asyncpg-stubs

import asyncio import asyncpg import os async def main(): # asyncpg-stubs provides type hints for asyncpg objects like Connection and Record. # Replace with your PostgreSQL connection string or use environment variables. # Example: DATABASE_URL='postgresql://user:password@localhost:5432/testdb' conn_string = os.environ.get('DATABASE_URL', 'postgresql://user:password@localhost:5432/testdb') conn: asyncpg.Connection | None = None try: conn = await asyncpg.connect(conn_string) print("Connected to PostgreSQL database.") # Execute a simple query result: str = await conn.fetchval("SELECT 'Hello from asyncpg!'") print(f"Query result: {result}") # Example with parameters and fetching a record await conn.execute(''' CREATE TABLE IF NOT EXISTS my_table ( id serial PRIMARY KEY, name text ); ''') await conn.execute("INSERT INTO my_table(name) VALUES($1)", "Test User") record: asyncpg.Record | None = await conn.fetchrow("SELECT id, name FROM my_table WHERE name = $1", "Test User") if record: print(f"Fetched record: ID={record['id']}, Name={record['name']}") except Exception as e: print(f"Error: {e}") finally: if conn: await conn.close() print("Connection closed.") if __name__ == "__main__": asyncio.run(main())