tessl/pypi-psycopg

PostgreSQL database adapter for Python

—

Pending

Overview

Eval results

Files

Type System and Adaptation

Name: tessl/pypi-psycopg
Author: tessl

PostgreSQL type system integration with automatic type conversion, custom type registration, and support for PostgreSQL-specific types like arrays, JSON, ranges, and geometric types.

Capabilities

Type Information Management

Core classes for managing PostgreSQL type metadata and registering type information.

class TypeInfo:
    """PostgreSQL type information container"""
    
    def __init__(
        self,
        name: str,
        oid: int,
        array_oid: int,
        *,
        regtype: str = "",
        delimiter: str = ",",
        typemod: type = None
    ):
        """
        Create type information object.
        
        Args:
            name: PostgreSQL type name
            oid: Type object identifier
            array_oid: Array type OID
            regtype: Type registration name (defaults to empty string)
            delimiter: Array element delimiter (defaults to comma)
            typemod: Type modifier class
        """
    
    @property
    def name(self) -> str:
        """PostgreSQL type name"""
    
    @property
    def oid(self) -> int:
        """Type object identifier"""
    
    @property
    def array_oid(self) -> int:
        """Array type OID (0 if no array type exists)"""
    
    @property
    def regtype(self) -> str | None:
        """Type registration name for PostgreSQL"""
    
    @property
    def delimiter(self) -> str:
        """Array element delimiter character"""
    
    @classmethod
    def fetch(
        cls,
        conn,
        name_or_oid: str | int
    ) -> TypeInfo:
        """
        Fetch type information from database.
        
        Args:
            conn: Database connection
            name_or_oid: Type name or OID to fetch
            
        Returns:
            TypeInfo object with database metadata
        """
    
    def register(self, context: AdaptContext) -> None:
        """
        Register this type in adaptation context.
        
        Args:
            context: Adaptation context to register in
        """

class TypesRegistry:
    """Registry for PostgreSQL type information and adapters"""
    
    def get_by_oid(self, oid: int) -> TypeInfo | None:
        """
        Get type information by OID.
        
        Args:
            oid: PostgreSQL type OID
            
        Returns:
            TypeInfo if found, None otherwise
        """
    
    def get_by_name(self, name: str) -> TypeInfo | None:
        """
        Get type information by name.
        
        Args:
            name: PostgreSQL type name
            
        Returns:
            TypeInfo if found, None otherwise
        """
    
    def add(self, type_info: TypeInfo) -> None:
        """
        Add type information to registry.
        
        Args:
            type_info: TypeInfo to add
        """

Adapter System

Core adaptation system for converting between Python objects and PostgreSQL data.

from typing import Any, Callable, Type

class AdaptersMap:
    """Map of type adapters for connections"""
    
    @property
    def types(self) -> TypesRegistry:
        """Associated types registry"""
    
    def register_loader(
        self,
        oid: int,
        loader: Callable[[bytes], Any]
    ) -> None:
        """
        Register function to load PostgreSQL data to Python.
        
        Args:
            oid: PostgreSQL type OID
            loader: Function to convert bytes to Python object
        """
    
    def register_dumper(
        self,
        cls: Type,
        dumper: Callable[[Any], bytes]
    ) -> None:
        """
        Register function to dump Python data to PostgreSQL.
        
        Args:
            cls: Python class to handle
            dumper: Function to convert Python object to bytes
        """
    
    def get_loader(self, oid: int) -> Callable[[bytes], Any] | None:
        """Get loader function for PostgreSQL type OID"""
    
    def get_dumper(self, obj: Any) -> Callable[[Any], bytes] | None:
        """Get dumper function for Python object"""

class Transformer:
    """Transform query parameters and results"""
    
    def __init__(self, context: AdaptContext):
        """
        Create transformer for adaptation context.
        
        Args:
            context: Adaptation context (connection or cursor)
        """
    
    def dump_sequence(
        self,
        params: Sequence[Any],
        format: PyFormat = PyFormat.AUTO
    ) -> list[bytes]:
        """
        Convert Python sequence to PostgreSQL format.
        
        Args:
            params: Python objects to convert
            format: Output format (text/binary)
            
        Returns:
            List of converted bytes objects
        """
    
    def load_sequence(
        self,
        row: Sequence[bytes],
        format: PyFormat = PyFormat.AUTO
    ) -> list[Any]:
        """
        Convert PostgreSQL row to Python objects.
        
        Args:
            row: Raw bytes from PostgreSQL
            format: Input format (text/binary)
            
        Returns:
            List of converted Python objects
        """

class Dumper:
    """Base class for converting Python objects to PostgreSQL format"""
    
    def __init__(self, cls: type, context: AdaptContext | None = None):
        """
        Initialize dumper for Python class.
        
        Args:
            cls: Python class this dumper handles
            context: Adaptation context (connection info)
        """
    
    def dump(self, obj: Any) -> bytes:
        """
        Convert Python object to PostgreSQL bytes.
        
        Args:
            obj: Python object to convert
            
        Returns:
            Object serialized as bytes for PostgreSQL
        """
    
    def quote(self, obj: Any) -> bytes:
        """
        Convert and quote Python object for SQL inclusion.
        
        Args:
            obj: Python object to quote
            
        Returns:
            SQL-safe quoted bytes representation
        """
    
    @property
    def oid(self) -> int:
        """PostgreSQL type OID this dumper produces"""

class Loader:
    """Base class for converting PostgreSQL data to Python objects"""
    
    def __init__(self, oid: int, context: AdaptContext | None = None):
        """
        Initialize loader for PostgreSQL type.
        
        Args:
            oid: PostgreSQL type OID this loader handles
            context: Adaptation context (connection info)
        """
    
    def load(self, data: bytes) -> Any:
        """
        Convert PostgreSQL bytes to Python object.
        
        Args:
            data: Raw bytes from PostgreSQL
            
        Returns:
            Converted Python object
        """

class AdaptContext:
    """Context for type adaptation with connection information"""
    
    def __init__(self, adapters: AdaptersMap | None = None):
        """
        Create adaptation context.
        
        Args:
            adapters: Adapters map to use (defaults to global)
        """
    
    @property
    def adapters(self) -> AdaptersMap:
        """Adapters map for this context"""
    
    @property
    def connection(self) -> Connection | None:
        """Associated connection (None for global context)"""
    
    @property
    def encoding(self) -> str:
        """Character encoding for text conversion"""

Built-in Type Support

Psycopg provides automatic adaptation for standard Python and PostgreSQL types.

Basic Types

# Automatic conversions (no explicit registration needed):

# Python -> PostgreSQL
None -> NULL
bool -> boolean  
int -> integer/bigint
float -> double precision
str -> text/varchar
bytes -> bytea
Decimal -> numeric
datetime.date -> date
datetime.time -> time
datetime.datetime -> timestamp
datetime.timedelta -> interval
uuid.UUID -> uuid

# PostgreSQL -> Python  
NULL -> None
boolean -> bool
integer/bigint -> int
double precision -> float
text/varchar -> str
bytea -> bytes
numeric -> Decimal
date -> datetime.date
time -> datetime.time  
timestamp -> datetime.datetime
interval -> datetime.timedelta
uuid -> uuid.UUID

Array Types

# Array support for all basic types
list[int] -> integer[]
list[str] -> text[]
list[datetime.date] -> date[]

# Multi-dimensional arrays
list[list[int]] -> integer[][]

# Custom array delimiter support via TypeInfo

JSON Types

# JSON/JSONB support
dict -> json/jsonb
list -> json/jsonb  
Any JSON-serializable object -> json/jsonb

# Custom JSON encoder/decoder registration

Custom Type Registration

def register_custom_type(
    conn,
    python_type: Type,
    pg_type_name: str,
    dumper: Callable[[Any], str],
    loader: Callable[[str], Any]
) -> None:
    """
    Register custom type conversion.
    
    Args:
        conn: Database connection
        python_type: Python class to register
        pg_type_name: PostgreSQL type name
        dumper: Function to convert Python object to string
        loader: Function to convert string to Python object
    """
    
    # Fetch type info from database
    type_info = TypeInfo.fetch(conn, pg_type_name)
    
    # Register dumper for Python -> PostgreSQL
    conn.adapters.register_dumper(python_type, dumper)
    
    # Register loader for PostgreSQL -> Python
    conn.adapters.register_loader(type_info.oid, loader)

Specialized Type Modules

Psycopg includes specialized modules for PostgreSQL-specific data types.

Array Types

from psycopg.types import array

# Array type registration
def register_array_type(
    conn,
    element_type_name: str,
    array_type_name: str
) -> None:
    """Register PostgreSQL array type for automatic handling"""

# Custom array handling
class ArrayDumper:
    """Custom array dumper for specialized formatting"""
    
class ArrayLoader:  
    """Custom array loader for specialized parsing"""

Range Types

from psycopg.types import range

class Range:
    """PostgreSQL range type representation"""
    
    def __init__(
        self, 
        lower=None, 
        upper=None, 
        bounds="[)", 
        empty=False
    ):
        """
        Create range object.
        
        Args:
            lower: Lower bound value
            upper: Upper bound value  
            bounds: Bound inclusion ("[)", "(]", "[]", "()")
            empty: True for empty range
        """
    
    @property
    def lower(self) -> Any:
        """Lower bound value"""
    
    @property
    def upper(self) -> Any:
        """Upper bound value"""
    
    @property
    def bounds(self) -> str:
        """Bound inclusion string"""
    
    @property
    def empty(self) -> bool:
        """True if range is empty"""

# Built-in range types
DateRange = Range[datetime.date]
TimestampRange = Range[datetime.datetime]  
NumericRange = Range[Decimal]

Composite Types

from psycopg.types import composite

def register_composite_type(
    conn,
    type_name: str,
    python_class: Type
) -> None:
    """
    Register PostgreSQL composite type with Python class.
    
    Args:
        conn: Database connection
        type_name: PostgreSQL composite type name
        python_class: Python class to map to
    """

# Custom composite type handling
class CompositeLoader:
    """Load PostgreSQL composite types to Python objects"""
    
class CompositeDumper:
    """Dump Python objects to PostgreSQL composite types"""

Enumeration Types

from psycopg.types import enum

def register_enum_type(
    conn,
    enum_type_name: str,
    python_enum: Type[Enum]
) -> None:
    """
    Register PostgreSQL enum type with Python Enum.
    
    Args:
        conn: Database connection  
        enum_type_name: PostgreSQL enum type name
        python_enum: Python Enum class
    """

# Usage with Python enums
from enum import Enum

class Color(Enum):
    RED = "red"
    GREEN = "green" 
    BLUE = "blue"

register_enum_type(conn, "color", Color)

Network Address Types

from psycopg.types import net
import ipaddress

# Automatic conversion
ipaddress.IPv4Address -> inet
ipaddress.IPv6Address -> inet  
ipaddress.IPv4Network -> cidr
ipaddress.IPv6Network -> cidr

UUID Types

import uuid

# Automatic UUID conversion
uuid.UUID -> uuid
# PostgreSQL uuid -> uuid.UUID

Geometric Types

from psycopg.types import geometry

# Point type
class Point:
    def __init__(self, x: float, y: float): ...
    
# Line, Circle, Box, Path, Polygon types available

Third-Party Integration

Integration modules for popular Python libraries.

NumPy Integration

from psycopg.types import numpy

# NumPy array <-> PostgreSQL array conversion
import numpy as np

# Automatic conversion when numpy is available
np.ndarray -> PostgreSQL array
PostgreSQL array -> np.ndarray

Shapely Integration

from psycopg.types import shapely

# Shapely geometry <-> PostGIS geometry conversion
from shapely.geometry import Point, LineString, Polygon

# Automatic conversion when shapely is available
Point -> PostGIS POINT
LineString -> PostGIS LINESTRING  
Polygon -> PostGIS POLYGON

Advanced Usage Examples

Custom Type Registration Example

from decimal import Decimal
from psycopg.types import TypeInfo

class Money:
    """Custom money type with currency"""
    def __init__(self, amount: Decimal, currency: str = "USD"):
        self.amount = amount
        self.currency = currency
    
    def __str__(self):
        return f"{self.amount} {self.currency}"

def register_money_type(conn):
    """Register custom money type"""
    
    # Create dumper (Python -> PostgreSQL)
    def dump_money(obj: Money) -> str:
        return f"{obj.amount}"  # PostgreSQL money type
    
    # Create loader (PostgreSQL -> Python)  
    def load_money(data: str) -> Money:
        # Parse PostgreSQL money format: $12.34
        amount_str = data.replace('$', '').replace(',', '')
        return Money(Decimal(amount_str))
    
    # Get type info from database
    money_info = TypeInfo.fetch(conn, "money")
    
    # Register adapters
    conn.adapters.register_dumper(Money, dump_money)
    conn.adapters.register_loader(money_info.oid, load_money)

# Usage
register_money_type(conn)

with conn.cursor() as cur:
    # Insert Money object
    cur.execute(
        "INSERT INTO products (name, price) VALUES (%s, %s)",
        ("Widget", Money(Decimal("19.99"), "USD"))
    )
    
    # Query returns Money objects
    cur.execute("SELECT name, price FROM products")
    for name, price in cur:
        print(f"{name}: {price}")  # Widget: 19.99 USD

Custom Array Type

from psycopg.types import array, TypeInfo

class Tag:
    """Simple tag object"""
    def __init__(self, name: str, color: str = "blue"):
        self.name = name
        self.color = color
    
    def __str__(self):
        return f"{self.name}:{self.color}"

def register_tag_array(conn):
    """Register tag array type"""
    
    # First register individual tag type
    def dump_tag(obj: Tag) -> str:
        return f"{obj.name}:{obj.color}"
    
    def load_tag(data: str) -> Tag:
        parts = data.split(":", 1)
        return Tag(parts[0], parts[1] if len(parts) > 1 else "blue")
    
    # Register tag type (assuming custom PostgreSQL type exists)
    tag_info = TypeInfo.fetch(conn, "tag")
    conn.adapters.register_dumper(Tag, dump_tag)
    conn.adapters.register_loader(tag_info.oid, load_tag)
    
    # Register tag array type
    tag_array_info = TypeInfo.fetch(conn, "_tag")  # Array type usually prefixed with _
    
    def dump_tag_array(obj_list: list[Tag]) -> str:
        return "{" + ",".join(dump_tag(tag) for tag in obj_list) + "}"
    
    def load_tag_array(data: str) -> list[Tag]:
        # Parse PostgreSQL array format
        inner = data[1:-1]  # Remove { }
        if not inner:
            return []
        return [load_tag(item.strip()) for item in inner.split(",")]
    
    conn.adapters.register_dumper(list[Tag], dump_tag_array)
    conn.adapters.register_loader(tag_array_info.oid, load_tag_array)

# Usage
register_tag_array(conn)

tags = [Tag("python", "green"), Tag("database", "blue"), Tag("api", "red")]

with conn.cursor() as cur:
    cur.execute(
        "INSERT INTO posts (title, tags) VALUES (%s, %s)",
        ("My Post", tags)
    )

Dynamic Type Discovery

def discover_custom_types(conn):
    """Discover and register all custom types in database"""
    
    with conn.cursor() as cur:
        # Find all custom types
        cur.execute("""
            SELECT typname, oid, typarray 
            FROM pg_type 
            WHERE typnamespace = (
                SELECT oid FROM pg_namespace WHERE nspname = 'public'
            )
            AND typtype = 'c'  -- Composite types
        """)
        
        for type_name, oid, array_oid in cur:
            print(f"Found custom type: {type_name} (OID: {oid})")
            
            # Register generic handler for composite types
            def load_composite(data: str, type_name=type_name):
                # Parse composite type format: (field1,field2,...)
                fields = data[1:-1].split(',')  # Remove ( )
                return {f"field_{i}": field.strip() for i, field in enumerate(fields)}
            
            conn.adapters.register_loader(oid, load_composite)
            
            if array_oid:
                # Register array version too
                def load_composite_array(data: str, loader=load_composite):
                    inner = data[1:-1]  # Remove { }
                    if not inner:
                        return []
                    items = []
                    # Parse composite array (more complex parsing needed for real use)
                    return [loader(item) for item in inner.split('","')]
                
                conn.adapters.register_loader(array_oid, load_composite_array)

# Auto-register all custom types
discover_custom_types(conn)

Performance Optimization

# Pre-compile dumpers/loaders for better performance
def create_optimized_adapters(conn):
    """Create optimized adapters for frequently used types"""
    
    import json
    from decimal import Decimal
    
    # Fast JSON dumper
    json_dumps = json.dumps
    def fast_json_dump(obj) -> str:
        return json_dumps(obj, separators=(',', ':'))  # No spaces
    
    # Fast decimal dumper  
    def fast_decimal_dump(obj: Decimal) -> str:
        return str(obj)
    
    # Register optimized dumpers
    conn.adapters.register_dumper(dict, fast_json_dump)
    conn.adapters.register_dumper(list, fast_json_dump)
    conn.adapters.register_dumper(Decimal, fast_decimal_dump)

Format Specifications

from enum import IntEnum

class PyFormat(IntEnum):
    """Query parameter and result format"""
    AUTO = 0      # Automatic format selection
    TEXT = 1      # Text format  
    BINARY = 2    # Binary format

Type System Constants

# Common PostgreSQL type OIDs
OID_BOOL = 16
OID_INT2 = 21  
OID_INT4 = 23
OID_INT8 = 20
OID_FLOAT4 = 700
OID_FLOAT8 = 701
OID_TEXT = 25
OID_BYTEA = 17
OID_DATE = 1082
OID_TIME = 1083
OID_TIMESTAMP = 1114
OID_TIMESTAMPTZ = 1184
OID_INTERVAL = 1186
OID_NUMERIC = 1700
OID_UUID = 2950
OID_JSON = 114
OID_JSONB = 3802

Install with Tessl CLI