aioodbc

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