tessl/pypi-psycopg2

Python-PostgreSQL Database Adapter

—

Pending

Overview

Eval results

Files

Advanced Cursor Types

Name: tessl/pypi-psycopg2
Author: tessl

Specialized cursor classes that enhance the standard cursor functionality by returning results as dictionaries, named tuples, or providing logging capabilities for development and debugging purposes.

Capabilities

Dictionary Cursors

Cursors that return rows as dictionary-like or real dictionary objects, providing attribute and key-based access to column values.

class DictConnection(connection):
    """Connection using DictCursor by default."""
    
    def cursor(self, *args, **kwargs):
        """Create DictCursor."""

class DictCursor(cursor):
    """Cursor returning dict-like rows."""
    
    def execute(self, query, vars=None):
        """Execute query and prepare dict-like results."""
    
    def callproc(self, procname, vars=None):
        """Call stored procedure with dict-like results."""
    
    def fetchone(self):
        """
        Fetch next row as DictRow.
        
        Returns:
        DictRow or None: Dictionary-like row object
        """
    
    def fetchmany(self, size=None):
        """
        Fetch multiple rows as DictRow objects.
        
        Returns:
        list[DictRow]: List of dictionary-like row objects
        """
    
    def fetchall(self):
        """
        Fetch all rows as DictRow objects.
        
        Returns:
        list[DictRow]: List of dictionary-like row objects
        """

class DictRow(list):
    """Dictionary-like row object."""
    
    def __getitem__(self, x):
        """Get item by index or column name."""
    
    def __setitem__(self, x, v):
        """Set item by index or column name."""
    
    def items(self):
        """Return (column_name, value) pairs."""
    
    def keys(self):
        """Return column names."""
    
    def values(self):
        """Return column values."""
    
    def get(self, x, default=None):
        """Get column value with default."""
    
    def copy(self):
        """Return OrderedDict copy."""

class RealDictConnection(connection):
    """Connection using RealDictCursor by default."""

class RealDictCursor(cursor):
    """Cursor with real dict rows."""
    
    def fetchone(self):
        """
        Fetch next row as real dictionary.
        
        Returns:
        dict or None: Real dictionary row
        """
    
    def fetchmany(self, size=None):
        """
        Fetch multiple rows as real dictionaries.
        
        Returns:
        list[dict]: List of dictionary rows
        """
    
    def fetchall(self):
        """
        Fetch all rows as real dictionaries.
        
        Returns:
        list[dict]: List of dictionary rows
        """

class RealDictRow(dict):
    """Real dictionary row object (OrderedDict)."""

Usage Example:

import psycopg2
from psycopg2.extras import DictCursor, RealDictCursor, DictConnection

# Using DictCursor
conn = psycopg2.connect("host=localhost dbname=mydb user=myuser")
with conn.cursor(cursor_factory=DictCursor) as cur:
    cur.execute("SELECT id, name, email FROM users WHERE id = %s", (1,))
    user = cur.fetchone()
    
    # Access by column name or index
    print(f"Name: {user['name']}")
    print(f"Email: {user[2]}")
    print(f"Keys: {list(user.keys())}")

# Using RealDictCursor for real dictionaries
with conn.cursor(cursor_factory=RealDictCursor) as cur:
    cur.execute("SELECT * FROM users LIMIT 3")
    users = cur.fetchall()
    
    for user in users:
        # Real dictionary access
        print(f"User: {user}")
        print(f"Name: {user.get('name', 'Unknown')}")

# Using DictConnection for default dict cursors
dict_conn = psycopg2.connect(
    "host=localhost dbname=mydb user=myuser",
    connection_factory=DictConnection
)
with dict_conn.cursor() as cur:
    cur.execute("SELECT name, age FROM users")
    for row in cur:
        print(f"{row['name']} is {row['age']} years old")

conn.close()
dict_conn.close()

Named Tuple Cursors

Cursors that return rows as named tuples, providing attribute access to column values with automatic type generation.

class NamedTupleConnection(connection):
    """Connection using NamedTupleCursor by default."""

class NamedTupleCursor(cursor):
    """Cursor returning named tuples."""
    
    def execute(self, query, vars=None):
        """Execute query and prepare named tuple results."""
    
    def executemany(self, query, vars):
        """Execute multiple times with named tuple results."""
    
    def callproc(self, procname, vars=None):
        """Call stored procedure with named tuple results."""
    
    def fetchone(self):
        """
        Fetch next row as named tuple.
        
        Returns:
        namedtuple or None: Named tuple row
        """
    
    def fetchmany(self, size=None):
        """
        Fetch multiple rows as named tuples.
        
        Returns:
        list[namedtuple]: List of named tuple rows
        """
    
    def fetchall(self):
        """
        Fetch all rows as named tuples.
        
        Returns:
        list[namedtuple]: List of named tuple rows
        """
    
    # Class attributes
    Record = None  # Generated namedtuple class for current query
    MAX_CACHE = 1024  # Maximum cached namedtuple classes

Usage Example:

import psycopg2
from psycopg2.extras import NamedTupleCursor

conn = psycopg2.connect("host=localhost dbname=mydb user=myuser")

with conn.cursor(cursor_factory=NamedTupleCursor) as cur:
    cur.execute("SELECT id, name, email, age FROM users WHERE age > %s", (25,))
    
    # Access by attribute name
    for user in cur:
        print(f"ID: {user.id}")
        print(f"Name: {user.name}")
        print(f"Email: {user.email}")
        print(f"Age: {user.age}")
    
    # The Record class is auto-generated
    print(f"Record type: {cur.Record}")
    print(f"Record fields: {cur.Record._fields}")

conn.close()

Logging Cursors

Cursors that log executed queries and their execution time, useful for development, debugging, and performance monitoring.

class LoggingConnection(connection):
    """Connection that logs queries."""
    
    def initialize(self, logobj):
        """
        Initialize logging with logger or file.
        
        Parameters:
        - logobj: Logger object or file-like object
        """
    
    def filter(self, msg, curs):
        """
        Filter log messages.
        
        Parameters:
        - msg (str): Log message
        - curs: Cursor object
        
        Returns:
        str: Filtered message
        """
    
    def log(self, msg, curs):
        """
        Log method (set by initialize).
        
        Parameters:
        - msg (str): Message to log
        - curs: Cursor object
        """

class LoggingCursor(cursor):
    """Cursor that logs executed queries."""
    
    def execute(self, query, vars=None):
        """Execute and log query."""
    
    def callproc(self, procname, vars=None):
        """Execute procedure and log."""

class MinTimeLoggingConnection(LoggingConnection):
    """Log queries over time threshold."""
    
    def initialize(self, logobj, mintime=0):
        """
        Initialize with time threshold.
        
        Parameters:
        - logobj: Logger object or file-like object
        - mintime (float): Minimum execution time to log (seconds)
        """
    
    def filter(self, msg, curs):
        """Filter by execution time."""

class MinTimeLoggingCursor(LoggingCursor):
    """Cursor for MinTimeLoggingConnection."""
    
    @property
    def timestamp(self):
        """Execution timestamp."""

Usage Example:

import psycopg2
import logging
from psycopg2.extras import LoggingConnection, MinTimeLoggingConnection

# Set up logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger('psycopg2')

# Basic query logging
conn = psycopg2.connect(
    "host=localhost dbname=mydb user=myuser",
    connection_factory=LoggingConnection
)
conn.initialize(logger)

with conn.cursor() as cur:
    cur.execute("SELECT * FROM users WHERE age > %s", (25,))
    users = cur.fetchall()
    # Logs: "SELECT * FROM users WHERE age > 25"

# Log only slow queries (> 0.1 seconds)
slow_conn = psycopg2.connect(
    "host=localhost dbname=mydb user=myuser", 
    connection_factory=MinTimeLoggingConnection
)
slow_conn.initialize(logger, mintime=0.1)

with slow_conn.cursor() as cur:
    cur.execute("SELECT COUNT(*) FROM large_table")
    # Only logs if query takes > 0.1 seconds

# Log to file
with open('queries.log', 'a') as logfile:
    file_conn = psycopg2.connect(
        "host=localhost dbname=mydb user=myuser",
        connection_factory=LoggingConnection
    )
    file_conn.initialize(logfile)
    
    with file_conn.cursor() as cur:
        cur.execute("INSERT INTO audit_log (action) VALUES (%s)", ("user_login",))
    
    file_conn.commit()
    file_conn.close()

conn.close()
slow_conn.close()

Cursor Factories

Custom cursor creation and configuration for specialized use cases.

Usage Example:

import psycopg2
from psycopg2.extras import DictCursor, LoggingCursor

# Custom cursor factory
def my_cursor_factory(conn):
    """Custom cursor combining dict and logging functionality."""
    class MyCustomCursor(DictCursor, LoggingCursor):
        pass
    return MyCustomCursor(conn)

# Use custom cursor factory
conn = psycopg2.connect(
    "host=localhost dbname=mydb user=myuser",
    cursor_factory=my_cursor_factory
)

# Set up logging for custom cursor
if hasattr(conn, 'initialize'):
    import logging
    logger = logging.getLogger('custom')
    conn.initialize(logger)

with conn.cursor() as cur:
    cur.execute("SELECT id, name FROM users")
    for row in cur:
        # Dictionary access + logging
        print(f"User: {row['name']}")

conn.close()

Types

Dictionary Row Types

class DictRow(list):
    """Dictionary-like row that extends list."""
    
    def __getitem__(self, x) -> Any:
        """Get by index (int) or column name (str)."""
    
    def __setitem__(self, x, v) -> None:
        """Set by index (int) or column name (str)."""
    
    def items(self) -> list[tuple[str, Any]]:
        """Column name, value pairs."""
    
    def keys(self) -> list[str]:
        """Column names."""
    
    def values(self) -> list[Any]:
        """Column values."""
    
    def get(self, x, default=None) -> Any:
        """Get column value with default."""

class RealDictRow(dict):
    """Real dictionary row (OrderedDict)."""

Named Tuple Cursor Attributes

class NamedTupleCursor:
    Record: type  # Auto-generated namedtuple class
    MAX_CACHE: int  # 1024 - Maximum cached namedtuple classes

Logging Connection Methods

class LoggingConnection:
    def initialize(self, logobj) -> None:
        """Initialize with logger or file object."""
    
    def filter(self, msg: str, curs) -> str:
        """Filter log messages."""
    
    def log(self, msg: str, curs) -> None:
        """Log method (configured by initialize)."""

class MinTimeLoggingConnection(LoggingConnection):
    def initialize(self, logobj, mintime: float = 0) -> None:
        """Initialize with time threshold."""

Install with Tessl CLI