Add AQ support

Author: GoodManWEN

README.md

@@ -18,14 +18,15 @@ Easy to use , buy may not the best practice for efficiency concern.
     pip install cx_Oracle_async
 
 ## Usage
-- Nearly all the same with aiomysql (with very limited functions of cource).
+- Nearly all the same with aiomysql (with very limited functions) , you can do execute , executemany , commit statement etc.
 - If you're connecting to database which is on a different machine from python process , you need to install oracle client module in order to use this library. Check [cx-Oracle's installation guide](https://cx-oracle.readthedocs.io/en/latest/user_guide/installation.html) for further information.
 - No automaticly date format transition built-in.
+- AQ feature newly added , [docs here](https://github.com/GoodManWEN/cx_Oracle_async/blob/main/docs/temporary_document_of_AQ.md).
 
 ## Performance
 query type | asynchronous multithreading | synchronous multithreading | synchronous single thread
 -|-|-|-
-fast single line query | 4864.96 q/s | 5859.20 q/s | 8209.536 q/s
+fast single line query | 6259.80 q/s | 28906.93 q/s | 14805.61 q/s
 single line insertion | N/A (todo) | N/A | N/A
 
 */\* Test platform: \*/*<br>
@@ -36,8 +37,8 @@ single line insertion | N/A (todo) | N/A | N/A
 
 ## Examples
 Before running examples , make sure you've already installed a [oracle client](https://github.com/GoodManWEN/cx_Oracle_async#usage) on your machine.
-```Python3
-# all_usages.py
+```Python
+# basic_usages.py
 import asyncio
 import cx_Oracle_async
 
@@ -80,23 +81,16 @@ if __name__ == '__main__':
 ```
 
 Or you can connect to database via makedsn style:
-```Python3
+```Python
 # makedsn.py
 import asyncio
 import cx_Oracle_async
 
 async def main():
     # same api as cx_Oracle.makedsn with 4 limited parameters(host , port , sid , service_name).
     dsn = cx_Oracle_async.makedsn(host = 'localhost' , port = '1521' , service_name = 'orcl')
-    oracle_pool = await cx_Oracle_async.create_pool(
-        user='user', 
-        password='password',
-        dsn = dsn
-    )
-
-    ...
-
-    await oracle_pool.close()
+    async with await cx_Oracle_async.create_pool(user='', password='',dsn = dsn) as pool:
+        ...
 
 asyncio.run(main())
 ```

cx_Oracle_async/AQ.py

@@ -0,0 +1,46 @@
+import cx_Oracle
+
+class AsyncQueueWrapper:
+    
+    def __init__(self , queue , loop , thread_pool , ):
+        self._queue = queue
+        self._loop = loop
+        self._thread_pool = thread_pool 
+
+    async def enqOne(self , *args , **kwargs):
+        return await self._loop.run_in_executor(self._thread_pool , self._queue.enqOne , *args , **kwargs)
+
+    async def enqMany(self , *args , **kwargs):
+        return await self._loop.run_in_executor(self._thread_pool , self._queue.enqMany , *args , **kwargs)
+
+    async def deqOne(self , *args , **kwargs):
+        return await self._loop.run_in_executor(self._thread_pool , self._queue.deqOne , *args , **kwargs)
+
+    def deqMany(self , maxMessages):
+        return DeqManyWrapper(self._loop , self._thread_pool , self._queue , maxMessages)
+
+class DeqManyWrapper:
+
+    def __init__(self , loop , thread_pool , queue , maxMessages):
+        self._loop = loop
+        self._thread_pool = thread_pool
+        self._queue = queue
+        self._count = 0
+        self._max = maxMessages
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        self._count += 1
+        if self._count <= self._max:
+            _tmp = self._queue.deqOptions.wait
+            self._queue.deqOptions.wait = cx_Oracle.DEQ_NO_WAIT
+            data = await self._loop.run_in_executor(self._thread_pool , self._queue.deqOne)
+            self._queue.deqOptions.wait = _tmp
+            if data:
+                return data
+            else:
+                raise StopAsyncIteration
+        else:
+            raise StopAsyncIteration

cx_Oracle_async/connections.py

@@ -0,0 +1,48 @@
+from .context import AbstractContextManager as BaseManager
+from .cursors import AsyncCursorWrapper , AsyncCursorWrapper_context
+from .AQ import AsyncQueueWrapper
+
+
+class AsyncConnectionWrapper_context(BaseManager):
+
+    def __init__(self , coro):
+        super().__init__(coro)
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self._obj.release()
+        self._obj = None
+
+
+class AsyncConnectionWrapper:
+
+    def __init__(self , conn , loop , thread_pool , pool):
+        self._conn = conn  
+        self._loop = loop
+        self._pool = pool
+        self._thread_pool = thread_pool
+        self.encoding = self._conn.encoding
+
+    def cursor(self):
+        coro = self._loop.run_in_executor(self._thread_pool , self._cursor)
+        return AsyncCursorWrapper_context(coro)
+
+    def _cursor(self):
+        return AsyncCursorWrapper(self._conn.cursor() , self._loop , self._thread_pool)
+
+    def msgproperties(self , *args , **kwargs):
+        return self._conn.msgproperties(*args , **kwargs)
+
+    async def queue(self , *args , **kwargs):
+        return AsyncQueueWrapper(self._conn.queue(*args , **kwargs) , self._loop , self._thread_pool)
+
+    async def gettype(self , *args , **kwargs):
+        '''
+        Uses the original cx_Oracle object without wrapper
+        '''
+        return await self._loop.run_in_executor(self._thread_pool , self._conn.gettype , *args , **kwargs)
+
+    async def commit(self):
+        await self._loop.run_in_executor(self._thread_pool , self._conn.commit)
+
+    async def release(self):
+        return await self._loop.run_in_executor(self._thread_pool , self._pool.release , self._conn)

cx_Oracle_async/context.py

@@ -0,0 +1,21 @@
+class AbstractContextManager:
+
+    def __init__(self , coro):
+        self._coro = coro
+        self._obj = None
+
+    def __next__(self):
+        return self.send(None)
+
+    def __iter__(self):
+        return self._coro.__await__()
+
+    def __await__(self):
+        return self._coro.__await__()
+
+    async def __aenter__(self):
+        self._obj = await self._coro
+        return self._obj
+
+    async def __aexit__(self, exc_type, exc, tb):
+        ...

cx_Oracle_async/cursors.py

@@ -0,0 +1,31 @@
+from .context import AbstractContextManager as BaseManager
+
+
+class AsyncCursorWrapper_context(BaseManager):
+
+    def __init__(self , coro):
+        super().__init__(coro)
+
+
+class AsyncCursorWrapper:
+
+    def __init__(self , cursor , loop , thread_pool):
+        self._cursor = cursor
+        self._loop = loop
+        self._thread_pool = thread_pool
+
+    async def execute(self , sql , *args , **kwargs):
+        return await self._loop.run_in_executor(self._thread_pool , self._cursor.execute , sql , *args , **kwargs)
+
+    async def executemany(self , sql , *args , **kwargs):
+        return await self._loop.run_in_executor(self._thread_pool , self._cursor.executemany , sql , *args , **kwargs)
+
+    async def fetchone(self):
+        return await self._loop.run_in_executor(self._thread_pool , self._cursor.fetchone)
+
+    async def fetchall(self):
+        # block mainly happens when fetch triggered.
+        return await self._loop.run_in_executor(self._thread_pool , self._cursor.fetchall)
+
+    async def var(self, args):
+        return await self._loop.run_in_executor(self._thread_pool , self._cursor.var, args)

cx_Oracle_async/pools.py

@@ -0,0 +1,46 @@
+from .context import AbstractContextManager as BaseManager
+from .connections import AsyncConnectionWrapper , AsyncConnectionWrapper_context
+from ThreadPoolExecutorPlus import ThreadPoolExecutor
+import asyncio
+import platform
+import os
+
+pltfm = platform.system()
+if pltfm == 'Windows':
+    DEFAULT_MAXIMUM_WORKER_NUM = (os.cpu_count() or 1) * 16
+    DEFAULT_MAXIMUM_WORKER_TIMES = 2
+elif pltfm == 'Linux' or pltfm == 'Darwin':
+    DEFAULT_MAXIMUM_WORKER_NUM = (os.cpu_count() or 1) * 32
+    DEFAULT_MAXIMUM_WORKER_TIMES = 3
+
+
+class AsyncPoolWrapper_context(BaseManager):
+
+    def __init__(self , coro):
+        super().__init__(coro)
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self._obj.close()
+        self._obj = None
+
+
+class AsyncPoolWrapper:
+    
+    def __init__(self , pool , loop = None):
+        
+        if loop == None:
+            loop = asyncio.get_running_loop()
+        self._thread_pool = ThreadPoolExecutor(max_workers = max(DEFAULT_MAXIMUM_WORKER_NUM , pool.max << DEFAULT_MAXIMUM_WORKER_TIMES)) 
+        self._thread_pool.set_daemon_opts(min_workers = max(4 , pool.min << 1))
+        self._loop = loop 
+        self._pool = pool 
+
+    def acquire(self):
+        coro = self._loop.run_in_executor(self._thread_pool , self._acquire)
+        return AsyncConnectionWrapper_context(coro)
+
+    def _acquire(self):
+        return AsyncConnectionWrapper(self._pool.acquire() , self._loop , self._thread_pool , self._pool)
+
+    async def close(self):
+        return await self._loop.run_in_executor(self._thread_pool , self._pool.close)