tessl/pypi-psycopg2-binary

PostgreSQL database adapter for Python with thread-safe connection pooling and SQL operations

—

Pending

Overview

Eval results

Files

Enhanced Cursors and Row Access

Name: tessl/pypi-psycopg2-binary
Author: tessl

Specialized cursor implementations providing dictionary-like row access, named tuple rows, real dictionary rows, and logging capabilities. These cursors extend the basic cursor functionality with convenient row access patterns for improved development experience.

Capabilities

Dictionary Cursors

Cursors that return rows as dictionary-like objects allowing access to columns by name instead of index position.

class DictCursorBase(cursor):
    """Base class for dictionary-like cursors."""
    
    def __init__(self, *args, **kwargs):
        """Initialize dict cursor base."""

class DictCursor(DictCursorBase):
    """Cursor returning DictRow objects."""
    
    def __init__(self, *args, **kwargs):
        """Initialize dictionary cursor."""
    
    def fetchone(self):
        """
        Fetch next row as DictRow.
        
        Returns:
        DictRow/None: Dictionary-like row object
        """
    
    def fetchmany(self, size=None):
        """
        Fetch multiple rows as DictRow objects.
        
        Returns:
        list: List of DictRow objects
        """
    
    def fetchall(self):
        """
        Fetch all rows as DictRow objects.
        
        Returns:
        list: List of DictRow objects
        """

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

class DictRow(list):
    """Dictionary-like row object."""
    
    def __init__(self, cursor):
        """Initialize from cursor description."""
    
    def __getitem__(self, x):
        """
        Get item by index or column name.
        
        Parameters:
        - x (int/str): Column index or name
        
        Returns:
        Value at column
        """
    
    def __setitem__(self, x, v):
        """
        Set item by index or column name.
        
        Parameters:
        - x (int/str): Column index or name
        - v: Value to set
        """
    
    def keys(self):
        """
        Get column names.
        
        Returns:
        iterator: Column names
        """
    
    def values(self):
        """
        Get column values.
        
        Returns:
        iterator: Column values
        """
    
    def items(self):
        """
        Get (name, value) pairs.
        
        Returns:
        iterator: (column_name, value) tuples
        """
    
    def get(self, x, default=None):
        """
        Get value with default.
        
        Parameters:
        - x (str): Column name
        - default: Default value if column not found
        
        Returns:
        Column value or default
        """
    
    def copy(self):
        """
        Create OrderedDict copy of row.
        
        Returns:
        OrderedDict: Copy of row data
        """
    
    def __contains__(self, x):
        """
        Check if column exists.
        
        Parameters:
        - x (str): Column name
        
        Returns:
        bool: True if column exists
        """

Usage examples:

from psycopg2.extras import DictCursor, DictConnection

# Using DictCursor with existing connection
cur = conn.cursor(cursor_factory=DictCursor)
cur.execute("SELECT id, name, email FROM users WHERE id = %s", (1,))
user = cur.fetchone()

print(user['name'])     # Access by column name
print(user[1])          # Access by index still works
print(user.get('age', 'Unknown'))  # Safe access with default

# Iterate over columns
for key, value in user.items():
    print(f"{key}: {value}")

# Using DictConnection (DictCursor by default)
dict_conn = psycopg2.connect(..., connection_factory=DictConnection)
cur = dict_conn.cursor()  # Automatically a DictCursor
cur.execute("SELECT * FROM products")
for row in cur:
    print(f"Product: {row['name']}, Price: {row['price']}")

Real Dictionary Cursors

Cursors that return actual Python dictionary objects instead of DictRow objects.

class RealDictCursor(DictCursorBase):
    """Cursor returning real dict objects."""
    
    def __init__(self, *args, **kwargs):
        """Initialize real dictionary cursor."""

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

class RealDictRow(dict):
    """Real dictionary row object."""
    
    def __init__(self, cursor):
        """Initialize from cursor description."""

Usage examples:

from psycopg2.extras import RealDictCursor

cur = conn.cursor(cursor_factory=RealDictCursor)
cur.execute("SELECT id, name, email FROM users")
users = cur.fetchall()

# Each row is a real Python dict
for user in users:
    print(user)  # {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}
    user['processed'] = True  # Can modify like normal dict

# Useful for JSON serialization
import json
json_users = json.dumps(users)  # Direct serialization

Named Tuple Cursors

Cursors that return rows as named tuples with column names as attributes.

class NamedTupleCursor(cursor):
    """Cursor returning named tuple objects."""
    
    def __init__(self, *args, **kwargs):
        """Initialize named tuple cursor."""

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

Usage examples:

from psycopg2.extras import NamedTupleCursor

cur = conn.cursor(cursor_factory=NamedTupleCursor)
cur.execute("SELECT id, name, email FROM users WHERE id = %s", (1,))
user = cur.fetchone()

# Access as attributes
print(user.name)    # Access by attribute name
print(user.email)   # Clean, readable access
print(user[0])      # Index access still works

# Named tuples are immutable and hashable
user_set = {user}   # Can be added to sets
# user.name = 'Bob'  # Would raise AttributeError

# Good for type hints and IDE support
def process_user(user_row):
    return f"Processing {user_row.name} ({user_row.email})"

Logging Cursors

Cursors that log SQL statements and execution times for debugging and monitoring.

class LoggingConnection(connection):
    """Connection with logging capabilities."""
    
    def __init__(self, *args, **kwargs):
        """Initialize logging connection."""
    
    def cursor(self, *args, **kwargs):
        """Create LoggingCursor."""

class LoggingCursor(cursor):
    """Cursor that logs SQL statements."""
    
    def __init__(self, *args, **kwargs):
        """Initialize logging cursor."""
    
    def execute(self, query, vars=None):
        """Execute with logging."""
    
    def executemany(self, query, vars_list):
        """Execute many with logging."""

class MinTimeLoggingConnection(LoggingConnection):
    """Connection with minimum time logging."""
    
    def __init__(self, *args, **kwargs):
        """
        Initialize with minimum logging time.
        
        Parameters:
        - mintime (float): Minimum execution time to log (seconds)
        """

class MinTimeLoggingCursor(LoggingCursor):
    """Cursor logging only slow queries."""
    
    def __init__(self, *args, **kwargs):
        """Initialize minimum time logging cursor."""

Usage examples:

import logging
from psycopg2.extras import LoggingConnection, MinTimeLoggingConnection

# Setup logging
logging.basicConfig(level=logging.DEBUG)

# Log all queries
log_conn = psycopg2.connect(..., connection_factory=LoggingConnection)
cur = log_conn.cursor()
cur.execute("SELECT * FROM users")  # Will be logged

# Log only slow queries (>0.5 seconds)
slow_conn = psycopg2.connect(..., 
    connection_factory=MinTimeLoggingConnection,
    mintime=0.5
)
cur = slow_conn.cursor()
cur.execute("SELECT * FROM large_table")  # Only logged if slow

Cursor Context Managers

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

# Automatic cursor cleanup
with conn.cursor(cursor_factory=DictCursor) as cur:
    cur.execute("SELECT * FROM users")
    users = cur.fetchall()
    # Cursor automatically closed

# Named cursor with context manager
with conn.cursor('server_cursor') as cur:
    cur.execute("SELECT * FROM large_table")
    for row in cur:
        process_row(row)
    # Server cursor automatically closed

Bulk Operations with Enhanced Cursors

Enhanced cursors work seamlessly with bulk operations for improved performance.

from psycopg2.extras import execute_batch, execute_values, RealDictCursor

# Bulk insert with real dict cursor (for result processing)
cur = conn.cursor(cursor_factory=RealDictCursor)
data = [
    {'name': 'Alice', 'email': 'alice@example.com'},
    {'name': 'Bob', 'email': 'bob@example.com'}
]

# Convert dicts to tuples for bulk insert
tuples = [(d['name'], d['email']) for d in data]
execute_batch(cur, 
    "INSERT INTO users (name, email) VALUES (%s, %s)",
    tuples
)

# Bulk select with dict cursor
cur.execute("SELECT * FROM users WHERE created_date = CURRENT_DATE")
new_users = cur.fetchall()  # List of dicts

# Process results easily
for user in new_users:
    send_welcome_email(user['email'])

Bulk Operations

Efficient bulk execution functions for improved performance with large data sets.

def execute_batch(cur, sql, argslist, page_size=100):
    """
    Execute statement in batches.
    
    Parameters:
    - cur (cursor): Database cursor
    - sql (str): SQL statement with placeholders
    - argslist (sequence): Sequence of parameter tuples
    - page_size (int): Number of statements per batch
    """

def execute_values(cur, sql, argslist, template=None, page_size=100, fetch=False):
    """
    Execute statement using VALUES syntax.
    
    Parameters:
    - cur (cursor): Database cursor
    - sql (str): SQL statement with VALUES placeholder
    - argslist (sequence): Sequence of parameter tuples
    - template (str, optional): VALUES template
    - page_size (int): Number of tuples per page
    - fetch (bool): Return results if True
    
    Returns:
    list/None: Query results if fetch=True
    """

Types

Row Factory Functions

def DictRowFactory(cursor):
    """Factory function for DictRow objects."""

def RealDictRowFactory(cursor):
    """Factory function for real dict objects."""

def NamedTupleRowFactory(cursor):
    """Factory function for named tuple objects."""

Cursor Factory Types

CursorFactory = callable[[connection], cursor]
RowFactory = callable[[cursor], Any]

Connection Factory Examples

# Custom connection with default cursor factory
class MyConnection(psycopg2.extensions.connection):
    def cursor(self, *args, **kwargs):
        kwargs.setdefault('cursor_factory', DictCursor)
        return super().cursor(*args, **kwargs)

# Use custom connection
conn = psycopg2.connect(..., connection_factory=MyConnection)
cur = conn.cursor()  # Automatically a DictCursor

Install with Tessl CLI