asyncpg-stubs

0.31.2 · active · verified Sat Apr 11

asyncpg-stubs is a Python package providing type stubs to enhance static type checking and type inference for the high-performance asynchronous PostgreSQL client library, asyncpg. Currently at version 0.31.2, it maintains a close versioning scheme with asyncpg (matching major.minor versions) and receives frequent updates to ensure compatibility and accuracy.

Warnings

Install

Imports

Quickstart

This example demonstrates a basic connection and query using `asyncpg`. When `asyncpg-stubs` is installed, type checkers (like MyPy or Pyright) will provide precise type hints for `asyncpg.Connection`, `asyncpg.Record`, and other asyncpg objects and methods used in this code, improving developer experience and catching potential type errors statically.

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

view raw JSON →