goodmanwen / cx_oracle_async

A very simple asynchronous wrapper that allows you to get access to the Oracle database in asyncio programs.

cx_Oracle_async

A very simple asynchronous wrapper that allows you to get access to the Oracle database in asyncio programs.

Easy to use , buy may not the best practice for efficiency concern.

About development release

In parallel with the main branch, we have worked on a development version based on SQLAlchemy's thread pool management facilities, please refer to Development notes for details.

Requirements

• cx_Oracle >= 8.1.0 (Take into consideration that author of cx_Oracle said he's trying to implement asyncio support , APIs maybe change in future version. Switch to 8.1.0 if there's something wrong makes it not gonna work.)
• ThreadPoolExecutorPlus >= 0.2.0

Install

pip install cx_Oracle_async

Feature

• Nearly all the same as aiomysql in asynchronous operational approach , with limited cx_Oracle feature support.
• No automaticly date format transition built-in.
• AQ feature added , check docs here for further information.
• You can modify some of the connection properties simply like you're using cx_Oracle.
• You can do basic insert / select / delete 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 for further information.

Documentation

https://cx_oracle_async.readthedocs.io

Performance

query type	asynchronous multithreading	synchronous multithreading	synchronous single thread
fast single line query	6259.80 q/s	28906.93 q/s	14805.61 q/s
single line insertion	1341.88 q/s	1898 q/s	1685.17 q/s

/* Test platform: */
AMD Ryzen 3700x
Windows 10 LTSC
Oracle 19c
You can find performance test codes here.

Examples

Before running examples , make sure you've already installed a Oracle Client on your machine.

# basic_usages.py
import asyncio
import cx_Oracle_async

async def main():
    oracle_pool = await cx_Oracle_async.create_pool(
        host='localhost', 
        port='1521',
        user='user', 
        password='password',
        service_name='orcl', 
        min = 2,
        max = 4,
    )

    async with oracle_pool.acquire() as connection:
        async with connection.cursor() as cursor:
            await cursor.execute("SELECT * FROM V$SESSION")
            print(await cursor.fetchall())

    await oracle_pool.close()

if __name__ == '__main__':
    asyncio.run(main())

cx_oracle_async's Issues

Support connection with dsn

Could you please add create_pool function with dsn parameter.
And it looks better to use cx_Oracle.makedsn to build dsn string instead of your variant "f"{host}:{port}/{db}"".
Example from documentation:
dsn = cx_Oracle.makedsn("dbhost.example.com", 1521, service_name="orclpdb1")
connection = cx_Oracle.connect("hr", userpwd, dsn, encoding="UTF-8")

Support AQ

Hello,
Could you please support AQ methods
https://cx-oracle.readthedocs.io/en/latest/user_guide/aq.html

How to get the result after update

when after execute, How to get the result after update

Create connection without pool

It it possible to use cx_Oracle_async without oracle pool and threads?

Async Close procedure for Oracle pool

Seems like the Close method of Oracle pool is missed for async.
I'm not shure is it just

    async def close_connection_async(self) -> None:
        if self._connection:
            await self._connection.close()

or should be like

    async def close_connection_async(self) -> None:
        if self._connection:
            self._connection.close()
            await self._connection.wait_closed()

Poetry SolverProblemError

When I add cx-Oracle-async by Poetry and call "poetry show" command I got the next error:
mydir>poetry show

SolverProblemError

Because myproject depends on cx-Oracle-async (^0.1.3) which doesn't match any versions, version solving failed.

at c:\users\myuser\appdata\local\programs\python\python39-32\lib\site-packages\poetry\puzzle\solver.py:241 in _solve
237│ packages = result.packages
238│ except OverrideNeeded as e:
239│ return self.solve_in_compatibility_mode(e.overrides, use_latest=use_latest)
240│ except SolveFailure as e:
→ 241│ raise SolverProblemError(e)
242│
243│ results = dict(
244│ depth_first_search(
245│ PackageNode(self._package, packages), aggregate_package_nodes

Consider using the new python-oracledb driver

Hi,
while I was wrangling with cx_Oracle and my M1, I found out that Oracle released a new way to connect to their database systems: https://oracle.github.io/python-oracledb/

Maybe that will make it easier to write the async functionality due to better integration with Python and the THIN approach.

使任意同步库快速变asyncio异步语法的方式 ，run_in_executor

https://www.cnblogs.com/ydf0509/p/14210083.html ，，参考这篇文章

AsyncCursorWarper add var

Please add
class AsyncCursorWarper:
...
async def var(self, args):
return await self._loop.run_in_executor(self._thread_pool , self._cursor.var, args)

And why "Warper", may be "Wrapper"?

Missing cursor attribute 'description'

Other python database packages (including cx_Oracle) include a method to get the column names which involves using cursor.description. It looks like the object 'description' is not available for cx_Oracle_async for the AsyncCursorWrapper. Is there a different method for returning the column names when using this package?

Cursor context manager is missing __aexit__

Describe the bug
Potential memory leak in cursors context manager.

To Reproduce
Call many times cursor.fetchone().

Gist with sample code: https://gist.github.com/ilosamart/272a7dadb639f19fd62b7947ae12ab5f
It contains three versions of the same method: one async (this lib), one async (this lib) which explicitly closes the cursor, one with pure cx_oracle.

Expected behavior
I expected that the cursor context manager closed the cursor.

Platform:

• OS: Ubuntu 20.04
• Oracle version: 12.2
• Python version 3.8

Additional context
I used the instructions on https://oracle.github.io/odpi/doc/user_guide/debugging.html to debug memory alloc inside OCI.

Default cx_Oracle parameters

Hello,

1. Sometimes I have got cx_Oracle.DatabaseError: ORA-24422
   I want to use FORCE parameter of SessionPool.close(force=False) for to make it true.
   https://cx-oracle.readthedocs.io/en/latest/api_manual/session_pool.html
   Could you please add this option.

2. cx_Oracle.SessionPool defaults is encoding=None, nencoding=None
   Could you please make it as original values.

When the number of async tasks awaiting acquire reaches ThreadPoolExecutorPlus max_workers, this program hangs

Describe the bug
When the number of async tasks awaiting acquire reaches ThreadPoolExecutorPlus max_workers, this program hangs.

To Reproduce
Run the following with the sixth line of code commented out (as-is) to hang.
Run the following with the sixth line of code uncommented to run successfully.
(sixth line has the # toggle commenting... comment on it)
(ThreadPoolExecutorPlus DEFAULT_MAXIMUM_WORKER_NUM is 32 * num-cores for Linux)

import cx_Oracle_async
import asyncio

dbspecs = dict(dsn="WFX742D", user="ars", password="ars")

max_pool_cnxs = 5
num_acquire_tasks = 32 * os.cpu_count() + max_pool_cnxs

#num_acquire_tasks -= 1 # toggle commenting of this line to hang/run
queue_len = 0

async def acquire_conn(pool, i):
   global queue_len
   queue_len += 1
   print_pool_info(f't-{i} acquire...', pool)
   conn = await pool.acquire()
   print_pool_info(f't-{i} postacquire', pool)
   queue_len -= 1
   conn._conn.begin()
   cursor = await conn.cursor()
   await cursor.execute('select current_timestamp from dual', [])
   rows = await cursor.fetchall()
   print(f"t-{i} rows: {len(rows)}")
   await conn.commit()
   await pool.release(conn)

async def test_many_threads():
   pool = await cx_Oracle_async.create_pool(**dbspecs, max=max_pool_cnxs)
   print_pool_info('init', pool)
   tasks = [];
   for i in range(0, num_acquire_tasks):
      tasks.append(asyncio.create_task(acquire_conn(pool, i), name=f'task-{i}'))
   await asyncio.wait(tasks)
   print_pool_info('done', pool)

def print_pool_info(action, pool):
   ipool = pool._pool
   print(f'{action} (queued {queue_len}, busy {ipool.busy}/{ipool.max})')

asyncio.run(test_many_threads())

Expected behavior
I expected the queries to be queued and run when connections became available.

Platform:

• OS: Red Hat Enterprise Linux Server release 7.9 (Maipo)
• Oracle version: 11.2.0.4
• Python version 3.8.11

Additional context
I worked around this by added a semaphore in my code to limit the number of acquires to the max cnxs.
What is the benefit of allowing more threads than cnxs?

Support Rollback and Ping

Could you please support
https://cx-oracle.readthedocs.io/en/latest/api_manual/connection.html
Connection.ping()
and
Connection.rollback()
methods.
This is must have.

Connection.module
Connection.action
Connection.client_identifier
Connection.clientinfo
Would be very nice too.

Thank you.

adding cursor.description and cursor.fetchmany()

i added a PR to patch this in #25 but incase somone lands up here and PR isnt accepted or delayed then a "sneaky" fix would be to do something like this:

 class Cursor():
        def __init__(self, baseObject):
            self.__class__ = type(baseObject.__class__.__name__,(self.__class__, baseObject.__class__),{})
            self.__dict__ = baseObject.__dict__

        @property
        def description(self):
            return self._cursor.description

        async def fetchmany(self, *args, **kwargs):
            return await self._loop.run_in_executor(self._thread_pool, self._cursor.fetchmany, *args, **kwargs)

and usage would be something like

conn = await self.connection.acquire()
self.cursor = Cursor(await conn.cursor())

(its less than ideal but hey... cowboy thursdays right!)

another issue for description: #19

create engine with sqlalchemy

Hi, thank's for you library!
Is it possible to use your library together with sqlalchemy in order to create an create_async_engine?
Thanks.