asyncpg-stubs
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
- breaking The `asyncpg-stubs` version should closely match the major and minor version of the `asyncpg` runtime library you are using. Mismatched versions can lead to incorrect type inference, type checking errors, or a lack of expected type information.
- gotcha Version `0.31.2` included a refactor to 'remove default type arguments'. If your codebase previously relied on implicit generic types that now require explicit specification for strict type checking, this change might flag new type errors.
- gotcha Older versions of `asyncpg-stubs` (e.g., prior to `0.31.1`) had less accurate type signatures for methods like `fetchmany()` and `execute()`, which could lead to type checking errors when using these methods.
Install
-
pip install asyncpg-stubs
Imports
- Connection
from asyncpg import Connection
- Pool
from asyncpg import Pool
- Record
from asyncpg import Record
Quickstart
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())