aioodbc

aioodbc is a Python 3.7+ 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. Internally aioodbc 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.

aioodbc 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.

Basic Example

aioodbc 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.

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())

Connection Pool

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

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())

Context Managers

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

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())

Installation

In a linux environment pyodbc (hence aioodbc) 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 aioodbc

Run tests

To run tests locally without docker, install unixodbc and sqlite driver:

$ sudo apt-get install unixodbc
$ sudo apt-get install libsqliteodbc

Create virtualenv and install package with requirements:

$ pip install -r requirements-dev.txt

Run tests, lints etc:

$ make fmt
$ make lint
$ make test

Other SQL Drivers

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

Requirements

• Python 3.7+
• pyodbc
• uvloop (optional)