tessl/pypi-asyncpg

An asyncio PostgreSQL driver for high-performance database connectivity with Python async/await syntax

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Prepared Statements

Name: tessl/pypi-asyncpg
Author: tessl

High-performance reusable query execution with server-side statement caching and optimized parameter binding. Prepared statements provide significant performance benefits for queries that are executed repeatedly with different parameters.

Capabilities

Statement Preparation

Creates a prepared statement on the server that can be executed multiple times with different parameters, eliminating parse overhead and enabling query plan caching.

async def prepare(
    self,
    query: str,
    *,
    name: str = None,
    timeout: float = None,
    record_class: type = None
) -> PreparedStatement

Parameters:

query: SQL query string with optional parameter placeholders ($1, $2, etc.)
name: Optional name for the prepared statement (auto-generated if not provided)
timeout: Statement preparation timeout in seconds
record_class: Custom record class for query results

Usage Example:

import asyncpg

async def example():
    conn = await asyncpg.connect('postgresql://user:pass@localhost/db')
    
    # Prepare a statement
    stmt = await conn.prepare('SELECT * FROM users WHERE age > $1 AND city = $2')
    
    # Execute with different parameters
    young_users = await stmt.fetch(18, 'New York')
    older_users = await stmt.fetch(65, 'Seattle')
    
    await conn.close()

Statement Execution

Execute prepared statements with various result formats, providing the same interface as direct connection methods but with improved performance.

async def execute(self, *args, timeout: float = None) -> str
async def fetch(self, *args, timeout: float = None, record_class: type = None) -> typing.List[Record]
async def fetchval(self, *args, column: int = 0, timeout: float = None) -> typing.Any
async def fetchrow(self, *args, timeout: float = None, record_class: type = None) -> typing.Optional[Record]
async def executemany(self, args, *, timeout: float = None) -> None

Usage Example:

# Single execution
result = await stmt.execute(25, 'Boston')
rows = await stmt.fetch(30, 'Chicago')
count = await stmt.fetchval(21, 'Portland')
single_row = await stmt.fetchrow(35, 'Denver')

# Batch execution
await stmt.executemany([
    (22, 'Austin'),
    (28, 'Miami'),
    (33, 'Atlanta')
])

Cursor Creation

Create cursors from prepared statements for efficient traversal of large result sets.

def cursor(
    self,
    *args,
    prefetch: int = None,
    timeout: float = None,
    record_class: type = None
) -> CursorFactory

Usage Example:

stmt = await conn.prepare('SELECT * FROM large_table WHERE category = $1')

async with stmt.cursor('electronics', prefetch=100) as cursor:
    async for row in cursor:
        process_row(row)

Statement Introspection

Access metadata about prepared statements including parameter types, result attributes, and statement details.

def get_name(self) -> str
def get_query(self) -> str
def get_statusmsg(self) -> str
def get_parameters(self) -> typing.Tuple[Type, ...]
def get_attributes(self) -> typing.Tuple[Attribute, ...]

Usage Example:

stmt = await conn.prepare('SELECT id, name, email FROM users WHERE age > $1')

print(f"Statement name: {stmt.get_name()}")
print(f"Query: {stmt.get_query()}")
print(f"Parameters: {[p.name for p in stmt.get_parameters()]}")
print(f"Result columns: {[a.name for a in stmt.get_attributes()]}")

Types

class PreparedStatement:
    """A prepared statement for efficient query execution."""
    
    def get_name(self) -> str
    def get_query(self) -> str
    def get_statusmsg(self) -> str
    def get_parameters(self) -> typing.Tuple[Type, ...]
    def get_attributes(self) -> typing.Tuple[Attribute, ...]
    
    async def execute(self, *args, timeout: float = None) -> str
    async def fetch(self, *args, timeout: float = None, record_class: type = None) -> typing.List[Record]
    async def fetchval(self, *args, column: int = 0, timeout: float = None) -> typing.Any
    async def fetchrow(self, *args, timeout: float = None, record_class: type = None) -> typing.Optional[Record]
    async def executemany(self, args, *, timeout: float = None) -> None
    
    def cursor(
        self,
        *args,
        prefetch: int = None,
        timeout: float = None,
        record_class: type = None
    ) -> CursorFactory

Performance Benefits

Prepared statements provide several performance advantages:

Parse Overhead Elimination: Query parsing is done once during preparation
Query Plan Caching: PostgreSQL can cache and reuse execution plans
Parameter Type Optimization: Parameter types are determined during preparation
Network Efficiency: Only parameter values are sent for subsequent executions
Memory Efficiency: Statement metadata is cached and reused

Best Practices

Use for Repeated Queries: Greatest benefit for queries executed multiple times
Parameter Validation: Prepared statements provide automatic type checking
Statement Caching: Connection maintains a cache of prepared statements
Resource Management: Prepared statements are automatically cleaned up when connection closes
Batch Operations: Use executemany() for bulk parameter variations