tessl/pypi-psycopg

PostgreSQL database adapter for Python

—

Pending

Overview

Eval results

Files

Query Execution and Cursors

Name: tessl/pypi-psycopg
Author: tessl

Comprehensive cursor functionality for query execution, result fetching, and data manipulation with multiple cursor types optimized for different performance and memory requirements.

Capabilities

Standard Cursors

Basic cursor functionality for query execution and result retrieval with full DB-API 2.0 compliance.

class Cursor:
    def execute(
        self,
        query,
        params=None,
        *,
        prepare: bool | None = None,
        binary: bool | None = None
    ) -> Cursor:
        """
        Execute a database query.
        
        Args:
            query: SQL query string or sql.Composable object
            params: Query parameters (tuple, list, or dict)
            prepare: Use prepared statements for better performance
            binary: Request binary format for results
            
        Returns:
            Self for method chaining
        """
        
    def executemany(
        self,
        query,
        params_seq,
        *,
        returning: bool = False
    ) -> None:
        """
        Execute query multiple times with different parameter sets.
        
        Args:
            query: SQL query string or sql.Composable object
            params_seq: Sequence of parameter sets
            returning: True if query returns results
        """
        
    def executescript(self, script: str) -> None:
        """Execute multiple SQL statements from a script"""
        
    def fetchone(self) -> Any:
        """
        Fetch next row from query results.
        
        Returns:
            Next row or None if no more rows
        """
        
    def fetchmany(self, size: int | None = None) -> list:
        """
        Fetch multiple rows from query results.
        
        Args:
            size: Number of rows to fetch (uses arraysize if None)
            
        Returns:
            List of rows (may be empty)
        """
        
    def fetchall(self) -> list:
        """
        Fetch all remaining rows from query results.
        
        Returns:
            List of all remaining rows
        """
        
    def close(self) -> None:
        """Close the cursor and free resources"""
        
    def scroll(self, value: int, mode: str = "relative") -> None:
        """
        Scroll cursor position in scrollable cursors.
        
        Args:
            value: Number of rows to scroll
            mode: 'relative' or 'absolute' positioning
        """

class AsyncCursor:
    async def execute(
        self,
        query,
        params=None,
        *,
        prepare: bool | None = None,
        binary: bool | None = None
    ) -> AsyncCursor:
        """Async version of Cursor.execute()"""
        
    async def executemany(
        self,
        query,
        params_seq,
        *,
        returning: bool = False
    ) -> None:
        """Async version of Cursor.executemany()"""
        
    async def fetchone(self) -> Any:
        """Async version of Cursor.fetchone()"""
        
    async def fetchmany(self, size: int | None = None) -> list:
        """Async version of Cursor.fetchmany()"""
        
    async def fetchall(self) -> list:
        """Async version of Cursor.fetchall()"""

Server-Side Cursors

Server-side cursors for memory-efficient processing of large result sets by keeping data on the PostgreSQL server.

class ServerCursor:
    def __init__(
        self,
        connection,
        name: str,
        *,
        scrollable: bool | None = None,
        withhold: bool = False
    ):
        """
        Create a server-side cursor.
        
        Args:
            connection: Database connection
            name: Cursor name (must be unique)
            scrollable: Enable bidirectional scrolling
            withhold: Keep cursor after transaction commit
        """
        
    @property
    def name(self) -> str:
        """Server-side cursor name"""
        
    @property
    def scrollable(self) -> bool | None:
        """True if cursor supports scrolling"""
        
    @property
    def withhold(self) -> bool:
        """True if cursor survives transaction commit"""
        
    def execute(self, query, params=None, *, binary: bool | None = None) -> ServerCursor:
        """Execute query using server-side cursor"""
        
    def executemany(self, query, params_seq) -> None:
        """Execute query multiple times using server cursor"""
        
    def fetchone(self) -> Any:
        """Fetch one row from server cursor"""
        
    def fetchmany(self, size: int | None = None) -> list:
        """Fetch multiple rows from server cursor"""
        
    def fetchall(self) -> list:
        """Fetch all remaining rows from server cursor"""

class AsyncServerCursor:
    # Async version of ServerCursor with same interface but async methods
    async def execute(self, query, params=None, *, binary: bool | None = None) -> AsyncServerCursor: ...
    async def fetchone(self) -> Any: ...
    async def fetchmany(self, size: int | None = None) -> list: ...
    async def fetchall(self) -> list: ...

Client-Side Cursors

Client-side cursors that fetch and buffer results locally for applications that need random access to result sets.

class ClientCursor:
    def execute(self, query, params=None, *, binary: bool | None = None) -> ClientCursor:
        """Execute query with client-side result buffering"""
        
    def fetchone(self) -> Any:
        """Fetch one row from client buffer"""
        
    def fetchmany(self, size: int | None = None) -> list:
        """Fetch multiple rows from client buffer"""
        
    def fetchall(self) -> list:
        """Return all buffered rows"""
        
    def scroll(self, value: int, mode: str = "relative") -> None:
        """Scroll within buffered results"""

class AsyncClientCursor:
    # Async version of ClientCursor
    async def execute(self, query, params=None, *, binary: bool | None = None) -> AsyncClientCursor: ...
    async def fetchone(self) -> Any: ...
    async def fetchmany(self, size: int | None = None) -> list: ...
    async def fetchall(self) -> list: ...

Raw Cursors

Low-level cursors that return raw bytes without type adaptation for maximum performance in specialized applications.

class RawCursor:
    """Cursor returning raw bytes without type conversion"""
    
    def execute(self, query, params=None, *, binary: bool | None = None) -> RawCursor:
        """Execute query returning raw bytes"""
        
    def fetchone(self) -> tuple[bytes, ...] | None:
        """Fetch one row as tuple of bytes"""
        
    def fetchmany(self, size: int | None = None) -> list[tuple[bytes, ...]]:
        """Fetch multiple rows as tuples of bytes"""
        
    def fetchall(self) -> list[tuple[bytes, ...]]:
        """Fetch all rows as tuples of bytes"""

class RawServerCursor:
    """Server-side cursor returning raw bytes"""
    # Same interface as RawCursor but server-side

class AsyncRawCursor:
    """Async cursor returning raw bytes"""
    async def execute(self, query, params=None, *, binary: bool | None = None) -> AsyncRawCursor: ...
    async def fetchone(self) -> tuple[bytes, ...] | None: ...
    async def fetchmany(self, size: int | None = None) -> list[tuple[bytes, ...]]: ...
    async def fetchall(self) -> list[tuple[bytes, ...]]: ...

class AsyncRawServerCursor:
    """Async server-side cursor returning raw bytes"""
    # Same interface as AsyncRawCursor but server-side

Cursor Properties and Metadata

Access cursor state, query information, and result metadata.

class Cursor:
    @property
    def description(self) -> list[Column] | None:
        """Column information for last query results"""
        
    @property
    def rowcount(self) -> int:
        """Number of rows affected by last query"""
        
    @property
    def arraysize(self) -> int:
        """Default number of rows for fetchmany()"""
        
    @arraysize.setter
    def arraysize(self, size: int) -> None:
        """Set default fetchmany() size"""
        
    @property
    def itersize(self) -> int:
        """Number of rows fetched per network round-trip"""
        
    @itersize.setter
    def itersize(self, size: int) -> None:
        """Set network fetch batch size"""
        
    @property
    def query(self) -> bytes | None:
        """Last executed query as bytes"""
        
    @property
    def params(self) -> Sequence | None:
        """Parameters used in last query"""
        
    @property
    def pgresult(self) -> PGresult | None:
        """Low-level PostgreSQL result object"""
        
    @property
    def connection(self) -> Connection:
        """Connection associated with this cursor"""
        
    @property
    def row_factory(self) -> RowFactory | None:
        """Current row factory for result formatting"""
        
    @row_factory.setter
    def row_factory(self, factory: RowFactory | None) -> None:
        """Set row factory for result formatting"""

Iterator Interface

Cursors support Python iterator protocol for convenient row-by-row processing.

class Cursor:
    def __iter__(self) -> Iterator:
        """Return iterator over query results"""
        
    def __next__(self) -> Any:
        """Get next row (used by iterator protocol)"""

# Usage example
for row in cursor:
    print(row)

Context Manager Support

All cursor types support context manager protocol for automatic resource cleanup.

class Cursor:
    def __enter__(self) -> Cursor:
        """Enter context manager"""
        
    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        """Exit context manager and close cursor"""

class AsyncCursor:
    async def __aenter__(self) -> AsyncCursor:
        """Enter async context manager"""
        
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        """Exit async context manager and close cursor"""

COPY Operations

Specialized cursor methods for high-performance bulk data operations using PostgreSQL's COPY protocol.

class Cursor:
    def copy(
        self,
        statement: str,
        params=None,
        *,
        writer: Callable | None = None
    ) -> Copy:
        """
        Start COPY operation for bulk data transfer.
        
        Args:
            statement: COPY SQL statement
            params: Query parameters
            writer: Custom data writer function
            
        Returns:
            Copy context manager
        """

class AsyncCursor:
    def copy(
        self,
        statement: str,
        params=None,
        *,
        writer: Callable | None = None
    ) -> AsyncCopy:
        """Async version of copy operation"""

Prepared Statements

Automatic prepared statement support for improved performance with repeated queries.

class Connection:
    def prepare(self, query: str) -> str:
        """
        Prepare a statement for repeated execution.
        
        Args:
            query: SQL query to prepare
            
        Returns:
            Prepared statement name
        """
        
    def prepared(self) -> dict[str, str]:
        """Get mapping of prepared statement names to queries"""

# Prepared statements are used automatically when prepare=True
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,), prepare=True)

Column Metadata

class Column:
    @property
    def name(self) -> str:
        """Column name"""
        
    @property
    def type_code(self) -> int:
        """PostgreSQL type OID"""
        
    @property
    def display_size(self) -> int | None:
        """Display size for column"""
        
    @property
    def internal_size(self) -> int | None:
        """Internal storage size"""
        
    @property
    def precision(self) -> int | None:
        """Numeric precision"""
        
    @property
    def scale(self) -> int | None:
        """Numeric scale"""
        
    @property
    def null_ok(self) -> bool | None:
        """True if column allows NULL values"""

Usage Examples

Basic Query Execution

# Simple query
with conn.cursor() as cur:
    cur.execute("SELECT version()")
    result = cur.fetchone()
    print(result)

# Parameterized query
with conn.cursor() as cur:
    cur.execute("SELECT * FROM users WHERE age > %s", (25,))
    users = cur.fetchall()
    for user in users:
        print(user)

Large Result Set Processing

# Server-side cursor for large results
with conn.cursor(name="large_query") as cur:
    cur.execute("SELECT * FROM huge_table")
    while True:
        rows = cur.fetchmany(1000)  # Fetch in batches
        if not rows:
            break
        process_batch(rows)

Bulk Operations

# Bulk insert
data = [("Alice", 30), ("Bob", 25), ("Charlie", 35)]
with conn.cursor() as cur:
    cur.executemany(
        "INSERT INTO users (name, age) VALUES (%s, %s)",
        data
    )

Async Operations

# Async query execution
async with conn.cursor() as cur:
    await cur.execute("SELECT * FROM users")
    async for row in cur:
        print(row)

Install with Tessl CLI

npx tessl i tessl/pypi-psycopg

docs

advanced-operations.md