aioodbc

aioodbc is a Python 3.5+ module that enables asynchronous access to ODBC databases using asyncio. It is built on top of the pyodbc library and maintains a similar interface and feel. To prevent blocking the asyncio event loop, aioodbc internally employs threads to handle the synchronous ODBC calls.

Key features include:

• An asyncio DBAPI-like interface for ODBC.
• Core objects: aioodbc.Connection, aioodbc.Cursor, and aioodbc.Pool.
• Support for connection pooling.

To install aioodbc, you must first ensure that pyodbc is available. On Linux environments, pyodbc (and consequently aioodbc) requires the unixODBC library.

1. Install unixodbc and unixodbc-dev using your package manager (e.g., apt-get).
2. Install aioodbc via pip.

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

When using aioodbc.create_pool, it is recommended to provide a ThreadPoolExecutor from the concurrent.futures module. By defining a dedicated executor for database work, you ensure that database operations use a specific set of worker threads. This prevents database tasks from competing with the application's default threads and allows the default thread pool to remain available for other tasks.

from concurrent.futures import ThreadPoolExecutor
import aioodbc

# Create a dedicated executor for database work
executor = ThreadPoolExecutor(max_workers=10)

# Pass the executor to create_pool
pool = await aioodbc.create_pool(dsn='...', executor=executor)

To use aioodbc, ensure your environment meets the following requirements:

• Python: 3.5 or higher (requires PEP 492 coroutines).
• pyodbc: The underlying driver interface.
• unixODBC: The ODBC driver manager (for Unix-like systems).

When calling aioodbc.connect, you can pass an asynchronous unary function to the after_created parameter. This function is executed immediately after the underlying pyodbc connection is established, allowing you to configure specific connection attributes such as .setencoding or .setdecoding on the raw connection object.

# Example concept: passing a function to configure the connection
async def configure_connection(connection):
    connection.setencoding(encoding='utf-8')
    connection.setdecoding(errors='replace')

conn = await aioodbc.connect(dsn='...', after_created=configure_connection)

For efficient connection management, use aioodbc.create_pool(). You can acquire connections from the pool using pool.acquire(). When using a pool, remember to close the pool and wait for it to close completely using pool.close() and await pool.wait_closed().

import asyncio
import aioodbc

async def test_pool():
    dsn = "Driver=SQLite3;Database=sqlite.db"
    pool = await aioodbc.create_pool(dsn=dsn)

    async with pool.acquire() as conn:
        cur = await conn.cursor()
        await cur.execute("SELECT 42;")
        r = await cur.fetchall()
        print(r)
        await cur.close()
        await conn.close()
    pool.close()
    await pool.wait_closed()

asyncio.run(test_pool())

The most idiomatic way to use aioodbc is through asynchronous context managers (async with). This ensures that resources like connection pools, individual connections, and cursors are properly closed even if an error occurs.

import asyncio
import aioodbc

async def main():
    dsn = 'Driver={SQLite3};Database=test.db;'  # Example DSN
    # Using context managers for Pool, Connection, and Cursor
    async with aioodbc.create_pool(dsn=dsn) as pool:
        async with pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute('SELECT 1')
                res = await cur.fetchone()
                print(res)

asyncio.run(main())

To perform a simple query, you can use aioodbc.connect() to establish a connection and then create a cursor to execute SQL commands. This example demonstrates a basic query that returns a single value.

import asyncio
import aioodbc

async def main():
    dsn = 'Driver={SQLite3};Database=test.db;'  # Example DSN
    async with aioodbc.connect(dsn=dsn) as conn:
        async with conn.cursor() as cur:
            await cur.execute('SELECT 42')
            res = await cur.fetchone()
            print(res)

asyncio.run(main())

Pool, Connection, and Cursor objects all support the asynchronous context management protocol (async with). This is the recommended way to ensure resources are properly cleaned up.

import asyncio
import aioodbc

async def test_example():
    dsn = "Driver=SQLite;Database=sqlite.db"

    async with aioodbc.create_pool(dsn=dsn) as pool:
        async with pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute("SELECT 42 AS age;")
                val = await cur.fetchone()
                print(val)
                print(val.age)

asyncio.run(test_example())

aioodbc provides an API that is nearly identical to pyodbc. The primary difference is that you must use await or yield from for method calls that perform I/O. Properties and attribute access (like accessing column names on a row object) remain unchanged.

import asyncio
import aioodbc

async def test_example():
    dsn = "Driver=SQLite;Database=sqlite.db"
    conn = await aioodbc.connect(dsn=dsn)

    cur = await conn.cursor()
    await cur.execute("SELECT 42 AS age;")
    rows = await cur.fetchall()
    print(rows)
    print(rows[0])
    print(rows[0].age)
    await cur.close()
    await conn.close()

asyncio.run(test_example())

For applications requiring multiple connections, use aioodbc.create_pool(). This allows you to manage a set of connections efficiently and acquire them as needed from the pool.

import asyncio
import aioodbc

async def main():
    dsn = 'Driver={SQLite3};Database=test.db;'  # Example DSN
    async with aioodbc.create_pool(dsn=dsn) as pool:
        async with pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute('SELECT 1')
                res = await cur.fetchone()
                print(res)

asyncio.run(main())

To use aioodbc, ensure your environment meets the following requirements:

• Python 3.7+
• pyodbc
• uvloop (optional, but supported and tested)