Skip to content

Commit 03f9db0

Browse files
GoodManWENGoodManWEN
committed
Update AQ
1 parent 5741caf commit 03f9db0

File tree

10 files changed

+312
-46
lines changed

10 files changed

+312
-46
lines changed

‎cx_Oracle_async/AQ.py‎

Lines changed: 36 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,21 @@
1-
import cx_Oracle
1+
from cx_Oracle import MessageProperties , DEQ_NO_WAIT
2+
from ThreadPoolExecutorPlus import ThreadPoolExecutor
23
from asyncio import Lock as aioLock
4+
from collections.abc import Iterable
5+
from typing import Union , TYPE_CHECKING
6+
if TYPE_CHECKING:
7+
from .connections import AsyncConnectionWrapper
8+
from cx_Oracle import Queue
9+
from asyncio.windows_events import ProactorEventLoop
310

411

512
class AsyncQueueWrapper:
613

7-
def __init__(self , queue , loop , thread_pool , ):
14+
def __init__(self , queue: 'Queue', loop: 'ProactorEventLoop' , thread_pool: ThreadPoolExecutor , conn : 'AsyncConnectionWrapper'):
815
self._queue = queue
916
self._loop = loop
1017
self._thread_pool = thread_pool
18+
self._conn = conn
1119
self._deqlock = aioLock()
1220

1321
async def enqOne(self , *args , **kwargs):
@@ -20,9 +28,22 @@ async def deqOne(self , *args , **kwargs):
2028
async with self._deqlock:
2129
return await self._loop.run_in_executor(self._thread_pool , self._queue.deqOne , *args , **kwargs)
2230

23-
def deqMany(self , maxMessages = -1):
31+
def deqMany(self , maxMessages: int = -1):
2432
return DeqManyWrapper(self._loop , self._thread_pool , self._queue , self._deqlock , maxMessages)
2533

34+
def _decode(self , _object: MessageProperties):
35+
return _object.payload.decode(self._conn.encoding)
36+
37+
def unpack(self , _object: Union[MessageProperties , list]):
38+
if isinstance(_object , Iterable):
39+
return list(map(self._decode , _object))
40+
else:
41+
return self._decode(_object)
42+
43+
@property
44+
def pack(self):
45+
return self._conn.msgproperties
46+
2647
@property
2748
def enqOptions(self):
2849
return self._queue.enqOptions
@@ -34,16 +55,24 @@ def deqOptions(self):
3455

3556
class DeqManyWrapper:
3657

37-
def __init__(self , loop , thread_pool , queue , deqlock ,maxMessages):
58+
def __init__(self , loop : 'ProactorEventLoop' , thread_pool : ThreadPoolExecutor, queue : 'Queue' , deqlock: aioLock , maxMessages : int):
3859
self._loop = loop
3960
self._thread_pool = thread_pool
4061
self._queue = queue
4162
self._count = 0
42-
self._max = maxMessages if maxMessages != -1 else (1 << 16 - 1)
63+
self._max = maxMessages if maxMessages > -1 else (1 << 16 - 1)
64+
self._max = self._max if self._max <= (1 << 16 - 1) else (1 << 16 - 1)
4365
self._deqlock = deqlock
4466

4567
def __await__(self):
46-
return self._loop.run_in_executor(self._thread_pool , self._queue.deqMany , self._max).__await__()
68+
yield from self._deqlock.acquire().__await__()
69+
try:
70+
ret = yield from self._loop.run_in_executor(self._thread_pool , self._queue.deqMany , self._max).__await__()
71+
except Exception as exc:
72+
raise exc
73+
finally:
74+
self._deqlock.release()
75+
return ret
4776

4877
def __aiter__(self):
4978
return self
@@ -53,7 +82,7 @@ async def __anext__(self):
5382
if self._count <= self._max:
5483
_tmp = self._queue.deqOptions.wait
5584
async with self._deqlock:
56-
self._queue.deqOptions.wait = cx_Oracle.DEQ_NO_WAIT
85+
self._queue.deqOptions.wait = DEQ_NO_WAIT
5786
data = await self._loop.run_in_executor(self._thread_pool , self._queue.deqOne)
5887
self._queue.deqOptions.wait = _tmp
5988
if data:

‎cx_Oracle_async/__init__.py‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,10 @@
11
__version__ = ''
22

3-
from .utils import create_pool , makedsn
3+
from .utils import create_pool , makedsn , DEQ_NO_WAIT , DEQ_WAIT_FOREVER
44

55
__all__ = (
66
'create_pool',
7-
'makedsn'
7+
'makedsn',
8+
'DEQ_NO_WAIT',
9+
'DEQ_WAIT_FOREVER'
810
)

‎cx_Oracle_async/connections.py‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,11 @@
11
from .context import AbstractContextManager as BaseManager
22
from .cursors import AsyncCursorWrapper , AsyncCursorWrapper_context
33
from .AQ import AsyncQueueWrapper
4-
4+
from cx_Oracle import Connection , SessionPool
5+
from ThreadPoolExecutorPlus import ThreadPoolExecutor
6+
from typing import TYPE_CHECKING
7+
if TYPE_CHECKING:
8+
from asyncio.windows_events import ProactorEventLoop
59

610
class AsyncConnectionWrapper_context(BaseManager):
711

@@ -15,12 +19,11 @@ async def __aexit__(self, exc_type, exc, tb):
1519

1620
class AsyncConnectionWrapper:
1721

18-
def __init__(self , conn , loop , thread_pool , pool):
22+
def __init__(self , conn: Connection, loop: 'ProactorEventLoop', thread_pool: ThreadPoolExecutor, pool: SessionPool):
1923
self._conn = conn
2024
self._loop = loop
2125
self._pool = pool
2226
self._thread_pool = thread_pool
23-
self.encoding = self._conn.encoding
2427

2528
def cursor(self):
2629
coro = self._loop.run_in_executor(self._thread_pool , self._cursor)
@@ -32,8 +35,12 @@ def _cursor(self):
3235
def msgproperties(self , *args , **kwargs):
3336
return self._conn.msgproperties(*args , **kwargs)
3437

38+
@property
39+
def encoding(self):
40+
return self._conn.encoding
41+
3542
async def queue(self , *args , **kwargs):
36-
return AsyncQueueWrapper(self._conn.queue(*args , **kwargs) , self._loop , self._thread_pool)
43+
return AsyncQueueWrapper(self._conn.queue(*args , **kwargs) , self._loop , self._thread_pool , self)
3744

3845
async def gettype(self , *args , **kwargs):
3946
'''

‎cx_Oracle_async/context.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
1+
from types import CoroutineType
2+
13
class AbstractContextManager:
24

3-
def __init__(self , coro):
5+
def __init__(self , coro : CoroutineType):
46
self._coro = coro
57
self._obj = None
68

‎cx_Oracle_async/cursors.py‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,21 @@
11
from .context import AbstractContextManager as BaseManager
2+
from ThreadPoolExecutorPlus import ThreadPoolExecutor
3+
from types import CoroutineType
4+
from cx_Oracle import Cursor
5+
from typing import TYPE_CHECKING
6+
if TYPE_CHECKING:
7+
from asyncio.windows_events import ProactorEventLoop
28

39

410
class AsyncCursorWrapper_context(BaseManager):
511

6-
def __init__(self , coro):
12+
def __init__(self , coro : CoroutineType):
713
super().__init__(coro)
814

915

1016
class AsyncCursorWrapper:
1117

12-
def __init__(self , cursor , loop , thread_pool):
18+
def __init__(self , cursor : Cursor, loop : 'ProactorEventLoop' , thread_pool : ThreadPoolExecutor):
1319
self._cursor = cursor
1420
self._loop = loop
1521
self._thread_pool = thread_pool

‎cx_Oracle_async/pools.py‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,14 @@
11
from .context import AbstractContextManager as BaseManager
22
from .connections import AsyncConnectionWrapper , AsyncConnectionWrapper_context
33
from ThreadPoolExecutorPlus import ThreadPoolExecutor
4+
from cx_Oracle import SessionPool
5+
from types import CoroutineType
46
import asyncio
57
import platform
68
import os
9+
from typing import TYPE_CHECKING
10+
if TYPE_CHECKING:
11+
from asyncio.windows_events import ProactorEventLoop
712

813
pltfm = platform.system()
914
if pltfm == 'Windows':
@@ -16,7 +21,7 @@
1621

1722
class AsyncPoolWrapper_context(BaseManager):
1823

19-
def __init__(self , coro):
24+
def __init__(self , coro : CoroutineType):
2025
super().__init__(coro)
2126

2227
async def __aexit__(self, exc_type, exc, tb):
@@ -26,8 +31,7 @@ async def __aexit__(self, exc_type, exc, tb):
2631

2732
class AsyncPoolWrapper:
2833

29-
def __init__(self , pool , loop = None):
30-
34+
def __init__(self , pool : SessionPool, loop : 'ProactorEventLoop' = None):
3135
if loop == None:
3236
loop = asyncio.get_running_loop()
3337
self._thread_pool = ThreadPoolExecutor(max_workers = max(DEFAULT_MAXIMUM_WORKER_NUM , pool.max << DEFAULT_MAXIMUM_WORKER_TIMES))

‎cx_Oracle_async/utils.py‎

Lines changed: 23 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,23 @@
1+
from .pools import AsyncPoolWrapper , AsyncPoolWrapper_context
12
from ThreadPoolExecutorPlus import ThreadPoolExecutor
23
import cx_Oracle as cxor
34
import asyncio
4-
from .pools import AsyncPoolWrapper , AsyncPoolWrapper_context
5+
from typing import TYPE_CHECKING
6+
if TYPE_CHECKING:
7+
from asyncio.windows_events import ProactorEventLoop
8+
59

610
makedsn = cxor.makedsn
11+
DEQ_NO_WAIT = cxor.DEQ_NO_WAIT
12+
DEQ_WAIT_FOREVER = cxor.DEQ_WAIT_FOREVER
713

814
async def _create_pool(
9-
host=None,
10-
port=None,
11-
service_name=None,
12-
sid=None,
13-
loop=None,
14-
dsn=None,
15+
host: str =None,
16+
port: str =None,
17+
service_name: str =None,
18+
sid: str =None,
19+
loop: 'ProactorEventLoop' =None,
20+
dsn: str =None,
1521
**kwargs
1622
):
1723
if loop == None:
@@ -27,11 +33,11 @@ async def _create_pool(
2733
return pool
2834

2935
def create_pool(
30-
user=None,
31-
password=None,
32-
dsn=None,
33-
min=2,
34-
max=4,
36+
user: str =None,
37+
password: str =None,
38+
dsn: str =None,
39+
min: int =2,
40+
max: int =4,
3541
increment=1,
3642
connectiontype=cxor.Connection,
3743
threaded=True,
@@ -46,11 +52,11 @@ def create_pool(
4652
maxLifetimeSession=0,
4753
sessionCallback=None,
4854
maxSessionsPerShard=0,
49-
host=None,
50-
port=None,
51-
service_name=None,
52-
sid=None,
53-
loop=None,
55+
host: str =None,
56+
port: str =None,
57+
service_name: str =None,
58+
sid: str =None,
59+
loop: 'ProactorEventLoop' =None,
5460
):
5561
coro = _create_pool(
5662
user=user,

0 commit comments

Comments
 (0)