Merge pull request #13 from GoodManWEN/dev

Dev

Author: GoodManWEN

.github/workflows/python-package.yml

@@ -47,7 +47,7 @@ jobs:
         run: |  # sleep to make sure Oracle server is fully loaded
           sleep 60            
           python -m pip install --upgrade pip
-          python -m pip install flake8 pytest pytest-asyncio
+          python -m pip install flake8 pytest pytest-asyncio async_timeout
           pip install -r requirements.txt
       - name: Lint with flake8
         run: |

README.md

@@ -91,7 +91,7 @@ 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')
-    async with await cx_Oracle_async.create_pool(user='', password='',dsn = dsn) as pool:
+    async with cx_Oracle_async.create_pool(user='', password='',dsn = dsn) as pool:
         ...
 
 asyncio.run(main())

cx_Oracle_async/connections.py

@@ -6,6 +6,7 @@
 from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from asyncio.windows_events import ProactorEventLoop
+    from .pools import AsyncPoolWrapper
 
 class AsyncConnectionWrapper_context(BaseManager):
 
@@ -19,12 +20,14 @@ async def __aexit__(self, exc_type, exc, tb):
 
 class AsyncConnectionWrapper:
 
-    def __init__(self , conn: Connection, loop: 'ProactorEventLoop', thread_pool: ThreadPoolExecutor, pool: SessionPool):
+    def __init__(self , conn: Connection, loop: 'ProactorEventLoop', thread_pool: ThreadPoolExecutor, pool: SessionPool, pool_wrapper:'AsyncPoolWrapper'):
         self._conn = conn  
         self._loop = loop
         self._pool = pool
+        self._pool_wrapper = pool_wrapper
         self._thread_pool = thread_pool
 
+
     def cursor(self):
         coro = self._loop.run_in_executor(self._thread_pool , self._cursor)
         return AsyncCursorWrapper_context(coro)
@@ -39,6 +42,42 @@ def msgproperties(self , *args , **kwargs):
     def encoding(self):
         return self._conn.encoding
 
+    @property
+    def dsn(self):
+        return self._conn.dsn
+
+    @property 
+    def module(self):
+        return self._conn.module
+
+    @module.setter
+    def module(self , arg):
+        self._conn.module = arg
+
+    @property 
+    def action(self):
+        return self._conn.action
+
+    @action.setter
+    def action(self , arg):
+        self._conn.action = arg
+
+    @property 
+    def client_identifier(self):
+        return self._conn.client_identifier
+
+    @client_identifier.setter
+    def client_identifier(self , arg):
+        self._conn.client_identifier = arg
+
+    @property 
+    def clientinfo(self):
+        return self._conn.clientinfo
+
+    @clientinfo.setter
+    def clientinfo(self , arg):
+        self._conn.clientinfo = arg
+
     async def queue(self , *args , **kwargs):
         return AsyncQueueWrapper(self._conn.queue(*args , **kwargs) , self._loop , self._thread_pool , self)
 
@@ -49,7 +88,17 @@ async def gettype(self , *args , **kwargs):
         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)
+        return 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)
+        self._pool_wrapper._unoccupied(self._conn)
+        return await self._loop.run_in_executor(self._thread_pool , self._pool.release , self._conn)
+
+    async def cancel(self):
+        return await self._loop.run_in_executor(self._thread_pool , self._conn.cancel)
+
+    async def ping(self):
+        return await self._loop.run_in_executor(self._thread_pool , self._conn.ping)
+
+    async def rollback(self):
+        return await self._loop.run_in_executor(self._thread_pool , self._conn.rollback)

cx_Oracle_async/pools.py

@@ -1,7 +1,7 @@
 from .context import AbstractContextManager as BaseManager
 from .connections import AsyncConnectionWrapper , AsyncConnectionWrapper_context
 from ThreadPoolExecutorPlus import ThreadPoolExecutor
-from cx_Oracle import SessionPool
+from cx_Oracle import Connection , SessionPool
 from types import CoroutineType
 import asyncio
 import platform
@@ -38,13 +38,37 @@ def __init__(self , pool : SessionPool, loop : 'ProactorEventLoop' = None):
         self._thread_pool.set_daemon_opts(min_workers = max(4 , pool.min << 1))
         self._loop = loop 
         self._pool = pool 
+        self._occupied = set()
 
     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)
+        _conn = self._pool.acquire()
+        self._occupied.update((_conn , ))
+        return AsyncConnectionWrapper(_conn , self._loop , self._thread_pool , self._pool , self)
 
-    async def close(self):
-        return await self._loop.run_in_executor(self._thread_pool , self._pool.close)
+    def _unoccupied(self , obj: Connection):
+        self._occupied.remove(obj)
+
+    async def release(self , conn: Connection):
+        return await self._loop.run_in_executor(self._thread_pool , self._pool.release , conn)
+
+    async def drop(self , conn: Connection):
+        return await self._loop.run_in_executor(self._thread_pool , self._pool.drop , conn)
+
+    async def close(self , force: bool = False , interrupt: bool = False):
+        '''
+        WARNING: option `interrupt` will force cancel all running connections before close
+        the pool. This may cause fetching thread no response forever in some legacy version
+        of oracle database such as 11 or lower.
+
+        Do make sure this option works fine with your working enviornment.
+        '''
+        while self._occupied:
+            _conn = self._occupied.pop()
+            if interrupt:
+                await self._loop.run_in_executor(self._thread_pool , _conn.cancel)
+
+        return await self._loop.run_in_executor(self._thread_pool , self._pool.close , force)

tests/test_behavior.py

@@ -24,7 +24,7 @@ async def test_new_table():
             ret = await cursor.fetchone()
             assert ret 
             if ret[0] > 0:
-                await cursor.execute("DTOP TABLE DEPT")
+                await cursor.execute("DROP TABLE DEPT")
 
             sql = f"""
                 CREATE TABLE DEPT
@@ -78,4 +78,4 @@ async def test_usage():
 
     cursor1 = await conn1.cursor()
     async with conn1.cursor() as cursor2:
-        assert type(cursor1) is type(cursor2) 
+        assert type(cursor1) is type(cursor2)

tests/test_connection.py

@@ -3,6 +3,7 @@
 import pytest
 import asyncio
 from cx_Oracle_async import *
+import time
 
 @pytest.mark.asyncio
 async def test_different_connect_ways():
@@ -35,4 +36,51 @@ async def test_different_connect_ways():
 
     ret = await oracle_pool.close()
     assert ret == None
-    
+    
+@pytest.mark.asyncio
+async def test_properties():
+    INAQ = 0.5
+    dsn = makedsn(
+        host = 'localhost', 
+        port = '1521', 
+        sid = 'xe'
+    )
+    oracle_pool = await create_pool(
+        user = 'system',
+        password = 'oracle',
+        dsn = dsn
+    )
+
+    async with oracle_pool.acquire() as conn:
+        st_time = time.time()
+        conn.module = 'hello world'
+        conn.action = 'test_action'
+        conn.client_identifier = 'test_identifier'
+        conn.clientinfo = 'test_info'
+        ed_time = time.time()
+        assert (ed_time - st_time) <= INAQ
+
+        async with conn.cursor() as cursor:
+            await cursor.execute("SELECT SID, MODULE ,ACTION ,CLIENT_IDENTIFIER , CLIENT_INFO FROM V$SESSION WHERE MODULE='hello world'")
+            r = await cursor.fetchall()
+            assert len(r) == 1
+            _ , _module , _action , _ciden , _cinfo = r[0]
+            assert _module == 'hello world'
+            assert _action == 'test_action'
+            assert _ciden == 'test_identifier'
+            assert _cinfo == 'test_info'
+
+        # test no update 
+        st_time = time.time()
+        conn.module = 'hello world2'
+        conn.action = 'test_action2'
+        conn.client_identifier = 'test_identifier2'
+        conn.clientinfo = 'test_info2'
+        ed_time = time.time()
+        assert (ed_time - st_time) <= INAQ
+
+        conn2 = await oracle_pool.acquire()
+        async with conn2.cursor() as cursor:
+            await cursor.execute("SELECT SID, MODULE ,ACTION ,CLIENT_IDENTIFIER , CLIENT_INFO FROM V$SESSION WHERE MODULE='hello world2'")
+            r = await cursor.fetchall()
+            assert len(r) == 0

tests/test_drop.py

@@ -0,0 +1,75 @@
+import os , sys
+sys.path.append(os.getcwd())
+import pytest
+import asyncio
+import time
+from async_timeout import timeout
+from cx_Oracle_async import *
+import cx_Oracle
+import threading
+
+async def create_long_query(oracle_pool):
+    async with oracle_pool.acquire() as conn:
+        cursor = await conn.cursor()
+        try:
+            await cursor.execute("BEGIN  DBMS_LOCK.SLEEP(:a); END;",(10,))
+        except Exception as e:
+            assert isinstance(e , cx_Oracle.OperationalError)
+
+def create_long_query_sync(oracle_pool):
+    '''
+    use sync function in order to avoid pytest loop never stop bug.
+    '''
+    try:
+        conn = oracle_pool._pool.acquire()
+        cursor = conn.cursor()
+        cursor.execute("BEGIN  DBMS_LOCK.SLEEP(:a); END;",(10,))
+    except:
+        ...
+
+@pytest.mark.asyncio
+async def test_force_close():
+    loop = asyncio.get_running_loop()
+    dsn  = makedsn('localhost','1521',sid='xe')
+    INAQ = 0.5
+    oracle_pool = await create_pool(user='system',password='oracle',dsn=dsn)
+    loop.create_task(create_long_query(oracle_pool))
+    st_time = time.time()
+    await asyncio.sleep(2)
+    await oracle_pool.close(force = True , interrupt = False)
+    ed_time = time.time()
+    assert (10 - INAQ) <= (ed_time - st_time) <= (10 + INAQ)
+
+    # test occupy
+    oracle_pool = await create_pool(user='system',password='oracle',dsn=dsn,max=4)
+    conn = await oracle_pool.acquire()
+    conn = await oracle_pool.acquire()
+    assert len(oracle_pool._occupied) == 2
+    conn = await oracle_pool.acquire()
+    assert len(oracle_pool._occupied) == 3
+    st_time = time.time()
+    await asyncio.sleep(2)
+    async with timeout(2):
+        # no running task , return immediately
+        await oracle_pool.close(force = True , interrupt = False)
+    ed_time = time.time()
+    assert (2 - INAQ) <= (ed_time - st_time) <= (2 + INAQ)
+
+    # test interrupt
+    oracle_pool = await create_pool(user='system',password='oracle',dsn=dsn,max=4)
+    st_time = time.time()
+    t = threading.Thread(target = create_long_query_sync , args = (oracle_pool,))
+    t.setDaemon(True)
+    t.start()
+    await asyncio.sleep(2)
+    exception_flag = False
+    try:
+        async with timeout(2):
+            # no response forever
+            await oracle_pool.close(force = True , interrupt = True)
+    except Exception as e:
+        exception_flag = True
+        assert isinstance(e , asyncio.TimeoutError)
+    ed_time = time.time()
+    assert exception_flag
+    assert (4 - INAQ) <= (ed_time - st_time) <= (10 - INAQ)

tests/test_ping.py

@@ -0,0 +1,27 @@
+import os , sys
+sys.path.append(os.getcwd())
+import pytest
+import asyncio
+from cx_Oracle_async import *
+import cx_Oracle
+
+@pytest.mark.asyncio
+async def test_ping():
+    dsn = makedsn(
+        host = 'localhost', 
+        port = '1521', 
+        sid = 'xe'
+    )
+    async with create_pool(user = 'system',password = 'oracle',dsn = dsn) as oracle_pool:
+        conn = await oracle_pool.acquire()
+        r = await conn.ping()
+        assert r == None
+
+        await conn.release()
+        exception_flag = False
+        try:
+            await conn.ping()
+        except Exception as e:
+            exception_flag = True
+            assert isinstance(e , cx_Oracle.InterfaceError)
+        assert exception_flag

tests/test_rollback.py

@@ -0,0 +1,44 @@
+import os , sys
+sys.path.append(os.getcwd())
+import pytest
+import asyncio
+from cx_Oracle_async import *
+import cx_Oracle
+
+@pytest.mark.asyncio
+async def test_ping():
+    dsn = makedsn(
+        host = 'localhost', 
+        port = '1521', 
+        sid = 'xe'
+    )
+    async with create_pool(user = 'system',password = 'oracle',dsn = dsn) as oracle_pool:
+        async with oracle_pool.acquire() as conn:
+            async with conn.cursor() as cursor:
+                # check if dept exesits
+                await cursor.execute("SELECT COUNT(*) FROM USER_TABLES WHERE TABLE_NAME = UPPER(:a)" , ('DEPT' , ))
+                ret = await cursor.fetchone()
+                assert ret 
+                if ret[0] > 0:
+                    await cursor.execute("DROP TABLE DEPT")
+
+                sql = f"""
+                    CREATE TABLE DEPT
+                       (DEPTNO NUMBER(2) CONSTRAINT PK_DEPT PRIMARY KEY,
+                        DNAME VARCHAR2(14),
+                        LOC VARCHAR2(13) 
+                    )
+                """
+                await cursor.execute(sql)
+
+                await cursor.execute(f"INSERT INTO DEPT(DEPTNO) VALUES (:a)" , (10 , ))
+                await conn.rollback()
+                await cursor.execute(f"INSERT INTO DEPT(DEPTNO) VALUES (:a)" , (12 , ))
+                await conn.commit()
+
+        async with oracle_pool.acquire() as conn:
+            async with conn.cursor() as cursor:
+                await cursor.execute(f"SELECT * FROM DEPT")
+                r = await cursor.fetchall()
+                assert len(r) == 1
+                assert r[0][0] == 12