asyncodbc

asyncodbc is a Python 3.9+ module that makes it possible to access ODBC databases with asyncio. It relies on the awesome pyodbc library and preserves the same look and feel. asyncodbc was written using async/await syntax (PEP492) and only support Python that is not end-of-life(EOL). Internally asyncodbc employs threads to avoid blocking the event loop, threads are not that as bad as you think! Other drivers like motor use the same approach.

asyncodbc is fully compatible and tested with uvloop. Take a look at the test suite, all tests are executed with both the default event loop and uvloop.

Supported Databases

asyncodbc should work with all databases supported by pyodbc. But for now the library has been tested with: SQLite, MySQL and PostgreSQL. Feel free to add other databases to the test suite by submitting a PR.

Basic Example

asyncodbc is based on pyodbc and provides the same api, you just need to use yield from conn.f() or await conn.f() instead of conn.f()

Properties are unchanged, so conn.prop is correct as well as conn.prop = val.

importasyncioimportasyncodbcasyncdeftest_example():
dsn='Driver=SQLite;Database=sqlite.db'conn=awaitasyncodbc.connect(dsn=dsn, loop=loop)
cur=awaitconn.cursor()
awaitcur.execute("SELECT 42 AS age;")
rows=awaitcur.fetchall()
print(rows)
print(rows[0])
print(rows[0].age)
awaitcur.close()
awaitconn.close()
asyncio.run(test_example())

Connection Pool

Connection pooling is ported from aiopg and relies on PEP492 features:

importasyncioimportasyncodbcasyncdeftest_pool():
dsn='Driver=SQLite;Database=sqlite.db'pool=awaitasyncodbc.create_pool(dsn=dsn, loop=loop)
asyncwithpool.acquire() asconn:
cur=awaitconn.cursor()
awaitcur.execute("SELECT 42;")
r=awaitcur.fetchall()
print(r)
awaitcur.close()
awaitconn.close()
pool.close()
awaitpool.wait_closed()
asyncio.run(test_pool())

Context Managers

Pool, Connection and Cursor objects support the context management protocol:

importasyncioimportasyncodbcasyncdeftest_example():
dsn='Driver=SQLite;Database=sqlite.db'asyncwithasyncodbc.create_pool(dsn=dsn, loop=loop) aspool:
asyncwithpool.acquire() asconn:
asyncwithconn.cursor() ascur:
awaitcur.execute('SELECT 42 AS age;')
val=awaitcur.fetchone()
print(val)
print(val.age)
asyncio.run(test_example())

Installation

In a linux environment pyodbc (hence asyncodbc) requires the unixODBC library. You can install it using your package manager, for example:

$ sudo apt-get install unixodbc
$ sudo apt-get install unixodbc-dev

then:

pip install asyncodbc

Run tests

For testing purposes you need to install the test group requirements:

$ uv pip install -r pyproject.toml --group test -e .

Then just execute:

$ make test_mssql

NOTE: Running tests requires Python 3.9 or higher.

Other SQL Drivers

• asyncpg - asyncio client for PostgreSQL
• aiomysql - asyncio client form MySQL

Requirements

• Python 3.9+
• pyodbc
• uvloop (optional)