aioodbc

0.5.0 · active · verified Sun Apr 12

aioodbc is a Python 3.7+ module that enables asynchronous access to ODBC databases using `asyncio`. It acts as an asynchronous wrapper around the `pyodbc` library, maintaining a similar API. The library internally uses threads to prevent blocking the event loop, a common strategy for integrating synchronous I/O operations into asynchronous applications. The current version is 0.5.0, with ongoing development focused on Python version compatibility, dependency updates, and API enhancements.

Warnings

breaking The explicit `loop` parameter was removed from `aioodbc.connect()` and `aioodbc.create_pool()` in version 0.4.0. The library now relies on the `asyncio.get_running_loop()` to automatically determine the event loop.
Fix: Remove the `loop=asyncio.get_event_loop()` (or similar) argument from `aioodbc.connect()` and `aioodbc.create_pool()` calls. E.g., `await aioodbc.create_pool(dsn=dsn)` instead of `await aioodbc.create_pool(dsn=dsn, loop=loop)`.
breaking The return type of `Cursor.execute` was fixed in version 0.2.0. Previously, it incorrectly returned a `pyodbc.Cursor` instance. It now correctly returns the `aioodbc.Cursor` instance.
Fix: Ensure your code does not rely on `Cursor.execute` returning a `pyodbc.Cursor` object. If you assigned the result of `execute` to a new variable and expected `pyodbc` specific methods, this would now be an `aioodbc.Cursor`.
breaking Starting with version 0.5.0, aioodbc requires a minimal `pyodbc` version of 5.0.1. Older versions of `pyodbc` might lead to compatibility issues or errors.
Fix: Upgrade your `pyodbc` installation to version 5.0.1 or newer: `pip install --upgrade pyodbc`.
gotcha aioodbc (inheriting from pyodbc) does not support named placeholders (e.g., `:name`, `%(name)s`) in SQL queries. Only question mark (`?`) placeholders are supported for parameter substitution.
Fix: Always use `?` as placeholders for parameters in your SQL queries and pass a tuple or list of values to the `execute` method in the correct order. Example: `await cur.execute("INSERT INTO my_table (col1, col2) VALUES (?, ?);", (val1, val2))`.
gotcha Failing to explicitly close connections and cursors when not using context managers can lead to resource leaks and unclosed connection warnings, especially when errors occur.
Fix: Always use `async with` for `aioodbc.create_pool`, `pool.acquire`, and `conn.cursor` to ensure proper resource management and automatic closing/releasing even in the presence of exceptions. If not using context managers, ensure `await cur.close()` and `await conn.close()` are called, ideally in `finally` blocks.

Install

pip install aioodbc Install with pip

Imports

aioodbc
```
import aioodbc
```
The primary module for all aioodbc functionalities.

Quickstart

This quickstart demonstrates creating an asynchronous connection pool, acquiring a connection, executing a query, inserting data, and fetching results using context managers for proper resource handling. Replace the placeholder DSN with your actual ODBC connection string. It also shows a basic `CREATE TABLE` and `INSERT` operation, followed by a `SELECT`.

import asyncio
import aioodbc
import os

async def main():
    # Replace with your actual ODBC DSN
    # Example DSN for SQLite, replace 'sqlite.db' with your database file/path
    # For SQL Server: 'Driver={ODBC Driver 17 for SQL Server};Server=your_server;Database=your_db;UID=your_user;PWD=your_password'
    dsn = os.environ.get('ODBC_DSN', 'Driver=SQLite;Database=sqlite.db')

    try:
        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 answer;")
                    val = await cur.fetchone()
                    print(f"The answer is: {val.answer}")
                    await cur.execute("CREATE TABLE IF NOT EXISTS test_table (id INTEGER, name TEXT);")
                    await cur.execute("INSERT INTO test_table (id, name) VALUES (?, ?);", (1, 'Test User'))
                    await conn.commit() # Commit changes if autocommit is False (default for aioodbc connections)
                    await cur.execute("SELECT * FROM test_table;")
                    rows = await cur.fetchall()
                    print(f"Fetched data: {rows}")
    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == "__main__":
    # It's good practice to ensure a DSN is set for real applications
    # For this example, if ODBC_DSN is not set, it defaults to a SQLite in-memory DB.
    # For testing, you might need to ensure appropriate ODBC drivers are installed on your system.
    asyncio.run(main())

view raw JSON →