tessl/pypi-psycopg2

Python-PostgreSQL Database Adapter

—

Pending

Overview

Eval results

Files

Type Adaptation and Casting

Name: tessl/pypi-psycopg2
Author: tessl

Comprehensive type conversion system between Python and PostgreSQL types, including support for arrays, JSON, ranges, UUID, network types, and custom type registration for seamless data exchange.

Capabilities

Core Type Adaptation

Convert Python objects to SQL representations and PostgreSQL values back to Python objects.

def adapt(obj):
    """
    Adapt Python object to SQL.
    
    Parameters:
    - obj: Python object to adapt
    
    Returns:
    ISQLQuote: Adapted object with getquoted() method
    """

def register_adapter(type, adapter):
    """
    Register object adapter.
    
    Parameters:
    - type: Python type to adapt
    - adapter: Adapter class or function
    """

def new_type(oids, name, castfunc):
    """
    Create new typecaster.
    
    Parameters:
    - oids (sequence): PostgreSQL type OIDs
    - name (str): Type name
    - castfunc (callable): Casting function
    
    Returns:
    type: New typecaster object
    """

def new_array_type(oids, name, base_caster):
    """
    Create array typecaster.
    
    Parameters:
    - oids (sequence): Array type OIDs
    - name (str): Array type name  
    - base_caster: Base element typecaster
    
    Returns:
    type: Array typecaster
    """

def register_type(obj, scope=None):
    """
    Register typecaster.
    
    Parameters:
    - obj: Typecaster object
    - scope (connection/cursor, optional): Registration scope
    """

Usage Example:

import psycopg2
from psycopg2.extensions import adapt, register_adapter, new_type, register_type

# Basic adaptation
value = adapt("Hello World")
print(value.getquoted())  # b"'Hello World'"

number = adapt(42)
print(number.getquoted())  # b'42'

# Custom adapter for Python objects
class Point:
    def __init__(self, x, y):
        self.x = x
        self.y = y

class PointAdapter:
    def __init__(self, point):
        self.point = point
    
    def getquoted(self):
        return f"POINT({self.point.x}, {self.point.y})".encode()

# Register custom adapter
register_adapter(Point, PointAdapter)

# Now Point objects can be used in queries
conn = psycopg2.connect("host=localhost dbname=mydb user=myuser")
with conn.cursor() as cur:
    point = Point(10, 20)
    cur.execute("INSERT INTO locations (position) VALUES (%s)", (point,))

# Custom typecaster for PostgreSQL to Python
def parse_point(value, cur):
    if value is None:
        return None
    # Parse "POINT(x, y)" format
    coords = value.strip("POINT()").split(",")
    return Point(float(coords[0]), float(coords[1]))

# Register typecaster (assuming point type OID is 600)
point_type = new_type((600,), "POINT", parse_point)
register_type(point_type)

conn.close()

Built-in Adapter Classes

Standard adapters for common Python types to PostgreSQL representations.

class Binary:
    """Binary data adapter."""
    
    def __init__(self, obj):
        """Initialize with bytes-like object."""
    
    def getquoted(self):
        """Return quoted binary representation."""

class AsIs:
    """Pass-through adapter."""
    
    def __init__(self, obj):
        """Initialize with object."""
    
    def getquoted(self):
        """Return object as-is."""

class QuotedString:
    """Quoted string adapter."""
    
    def __init__(self, obj):
        """Initialize with string."""
    
    def getquoted(self):
        """Return quoted string."""

class Boolean:
    """Boolean adapter."""
    
    def __init__(self, obj):
        """Initialize with boolean."""
    
    def getquoted(self):
        """Return 't' or 'f'."""

class Float:
    """Float adapter."""
    
    def __init__(self, obj):
        """Initialize with float."""
    
    def getquoted(self):
        """Return float representation."""

class Int:
    """Integer adapter."""
    
    def __init__(self, obj):
        """Initialize with integer."""
    
    def getquoted(self):
        """Return integer representation."""

Usage Example:

from psycopg2.extensions import Binary, AsIs, QuotedString

# Binary data
binary_data = Binary(b'\x00\x01\x02\xff')
print(binary_data.getquoted())  # Escaped binary representation

# Raw SQL (dangerous - use carefully)
raw_sql = AsIs("NOW()")
cur.execute("INSERT INTO events (created_at) VALUES (%s)", (raw_sql,))

# Custom string quoting
custom_string = QuotedString("O'Reilly")
print(custom_string.getquoted())  # Properly escaped string

JSON and JSONB Support

Handle JSON data types with automatic serialization and deserialization.

class Json:
    """JSON adapter class."""
    
    def __init__(self, adapted, dumps=None):
        """
        Initialize JSON adapter.
        
        Parameters:
        - adapted: Python object to serialize
        - dumps (callable, optional): Custom JSON serializer
        """
    
    def dumps(self, obj):
        """Serialize object to JSON."""
    
    def getquoted(self):
        """Return quoted JSON representation."""

def register_json(conn_or_curs=None, globally=False, loads=None, oid=None, array_oid=None, name='json'):
    """
    Register JSON typecasters.
    
    Parameters:
    - conn_or_curs (connection/cursor, optional): Registration scope
    - globally (bool): Register globally
    - loads (callable, optional): Custom JSON deserializer
    - oid (int, optional): JSON type OID
    - array_oid (int, optional): JSON array type OID
    - name (str): Type name
    """

def register_default_json(conn_or_curs=None, globally=False, loads=None):
    """Register for PostgreSQL 9.2+ JSON."""

def register_default_jsonb(conn_or_curs=None, globally=False, loads=None):
    """Register for PostgreSQL 9.4+ JSONB."""

Usage Example:

import psycopg2
import json
from psycopg2.extras import Json, register_default_json, register_default_jsonb

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

# Register JSON support
register_default_json(conn)
register_default_jsonb(conn)

# Insert JSON data
data = {'name': 'Alice', 'age': 30, 'hobbies': ['reading', 'swimming']}

with conn.cursor() as cur:
    cur.execute(
        "INSERT INTO users (profile) VALUES (%s)",
        (Json(data),)
    )
    
    # Query JSON data  
    cur.execute("SELECT profile FROM users WHERE profile->>'name' = %s", ('Alice',))
    result = cur.fetchone()
    print(type(result[0]))  # dict (automatically deserialized)
    print(result[0]['hobbies'])  # ['reading', 'swimming']

# Custom JSON serialization
def custom_serializer(obj):
    """Custom JSON serializer with date handling."""
    if hasattr(obj, 'isoformat'):
        return obj.isoformat()
    return json.dumps(obj)

from datetime import datetime
timestamp_data = {'event': 'login', 'timestamp': datetime.now()}

with conn.cursor() as cur:
    cur.execute(
        "INSERT INTO events (data) VALUES (%s)",
        (Json(timestamp_data, dumps=custom_serializer),)
    )

conn.commit()
conn.close()

Range Types

Support for PostgreSQL range types with Python range objects.

class Range:
    """Base PostgreSQL range type."""
    
    def __init__(self, lower=None, upper=None, bounds='[)', empty=False):
        """
        Initialize range.
        
        Parameters:
        - lower: Lower bound value
        - upper: Upper bound value  
        - bounds (str): Bound inclusion ('[)', '(]', '[]', '()')
        - empty (bool): Empty range flag
        """
    
    @property
    def lower(self):
        """Lower bound value."""
    
    @property
    def upper(self):
        """Upper bound value."""
    
    @property
    def isempty(self):
        """True if range is empty."""
    
    @property
    def lower_inf(self):
        """True if lower bound is infinite."""
    
    @property
    def upper_inf(self):
        """True if upper bound is infinite."""
    
    @property
    def lower_inc(self):
        """True if lower bound is inclusive."""
    
    @property
    def upper_inc(self):
        """True if upper bound is inclusive."""
    
    def __contains__(self, x):
        """Check if value in range."""

class NumericRange(Range):
    """For numeric ranges (int4range, int8range, numrange)."""

class DateRange(Range):
    """For daterange."""

class DateTimeRange(Range):
    """For tsrange."""

class DateTimeTZRange(Range):
    """For tstzrange."""

def register_range(pgrange, pyrange, conn_or_curs, globally=False):
    """
    Register custom range type.
    
    Parameters:
    - pgrange (str): PostgreSQL range type name
    - pyrange: Python range class
    - conn_or_curs: Connection or cursor
    - globally (bool): Register globally
    """

Usage Example:

import psycopg2
from psycopg2.extras import NumericRange, DateRange, register_range
from datetime import date

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

# Numeric ranges
age_range = NumericRange(18, 65, bounds='[)')  # 18 <= age < 65
print(25 in age_range)  # True
print(70 in age_range)  # False

# Date ranges  
vacation_range = DateRange(date(2023, 7, 1), date(2023, 7, 15), bounds='[]')
print(vacation_range.lower)  # 2023-07-01
print(vacation_range.upper)  # 2023-07-15

with conn.cursor() as cur:
    # Insert range data
    cur.execute(
        "INSERT INTO employee_ages (age_range) VALUES (%s)",
        (age_range,)
    )
    
    cur.execute(
        "INSERT INTO vacations (dates) VALUES (%s)",
        (vacation_range,)
    )
    
    # Query with range operations
    cur.execute(
        "SELECT * FROM employee_ages WHERE age_range @> %s",
        (30,)  # Find ranges containing 30
    )
    results = cur.fetchall()

# Custom range type
class IPRange(NumericRange):
    """Custom IP address range."""
    pass

# Register custom range (assuming custom IP range type)
register_range('iprange', IPRange, conn)

conn.commit()
conn.close()

UUID Support

Handle UUID data types with Python's uuid module.

class UUID_adapter:
    """Adapter for uuid.UUID objects."""
    
    def __init__(self, uuid):
        """Initialize with UUID object."""
    
    def getquoted(self):
        """Return quoted UUID representation."""

def register_uuid(oids=None, conn_or_curs=None):
    """
    Register UUID type support.
    
    Parameters:
    - oids (sequence, optional): UUID type OIDs
    - conn_or_curs (connection/cursor, optional): Registration scope
    """

Usage Example:

import psycopg2
import uuid
from psycopg2.extras import register_uuid

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

# Generate and use UUIDs
user_id = uuid.uuid4()
session_id = uuid.uuid4()

with conn.cursor() as cur:
    cur.execute(
        "INSERT INTO users (id, session_id, name) VALUES (%s, %s, %s)",
        (user_id, session_id, "Alice")
    )
    
    # Query by UUID
    cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    user = cur.fetchone()
    print(f"User ID type: {type(user[0])}")  # <class 'uuid.UUID'>

conn.commit()
conn.close()

Network Address Types

Support for PostgreSQL inet and cidr types with Python's ipaddress module.

class Inet:
    """Wrapper for inet values."""
    
    def __init__(self, addr):
        """Initialize with address string."""
    
    def getquoted(self):
        """Return quoted inet representation."""

def register_inet(oid=None, conn_or_curs=None):
    """Register inet type (deprecated)."""

def register_ipaddress(conn_or_curs=None):
    """
    Register ipaddress module support for inet/cidr types.
    
    Parameters:
    - conn_or_curs (connection/cursor, optional): Registration scope
    """

def cast_interface(s, cur=None):
    """Cast to IPv4/IPv6Interface."""

def cast_network(s, cur=None):
    """Cast to IPv4/IPv6Network."""

def adapt_ipaddress(obj):
    """Adapt ipaddress objects to SQL."""

Usage Example:

import psycopg2
import ipaddress
from psycopg2.extras import register_ipaddress

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

# Use Python ipaddress objects
ip_addr = ipaddress.IPv4Address('192.168.1.1')
ip_network = ipaddress.IPv4Network('192.168.1.0/24')
ip_interface = ipaddress.IPv4Interface('192.168.1.1/24')

with conn.cursor() as cur:
    cur.execute(
        "INSERT INTO network_config (server_ip, subnet, interface) VALUES (%s, %s, %s)",
        (ip_addr, ip_network, ip_interface)
    )
    
    # Query network data
    cur.execute("SELECT server_ip, subnet FROM network_config")
    for row in cur.fetchall():
        print(f"IP: {row[0]}, Network: {row[1]}")
        print(f"Types: {type(row[0])}, {type(row[1])}")

conn.commit()
conn.close()

HStore Support

Handle PostgreSQL hstore data type as Python dictionaries.

class HstoreAdapter:
    """Adapter for dict to hstore."""
    
    def __init__(self, wrapped):
        """Initialize with dictionary."""
    
    def getquoted(self):
        """Return quoted hstore representation."""

def register_hstore(conn_or_curs, globally=False, unicode=False, oid=None, array_oid=None):
    """
    Register hstore support.
    
    Parameters:
    - conn_or_curs: Connection or cursor
    - globally (bool): Register globally
    - unicode (bool): Use unicode strings
    - oid (int, optional): HStore type OID
    - array_oid (int, optional): HStore array type OID
    """

Usage Example:

import psycopg2
from psycopg2.extras import register_hstore

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

# Enable hstore extension in PostgreSQL first:
# CREATE EXTENSION IF NOT EXISTS hstore;

register_hstore(conn)

# Use Python dictionaries as hstore
metadata = {
    'color': 'blue',
    'size': 'large', 
    'material': 'cotton'
}

with conn.cursor() as cur:
    cur.execute(
        "INSERT INTO products (name, attributes) VALUES (%s, %s)",
        ("T-Shirt", metadata)
    )
    
    # Query hstore data
    cur.execute("SELECT attributes FROM products WHERE name = %s", ("T-Shirt",))
    result = cur.fetchone()
    print(type(result[0]))  # dict
    print(result[0]['color'])  # 'blue'
    
    # HStore operations
    cur.execute(
        "SELECT * FROM products WHERE attributes -> %s = %s",
        ('color', 'blue')
    )

conn.commit()
conn.close()

Composite Type Support

Handle PostgreSQL composite types as Python objects.

class CompositeCaster:
    """Caster for composite types."""
    
    def __init__(self, name, oid, attrs, array_oid=None, schema=None):
        """
        Initialize composite caster.
        
        Parameters:
        - name (str): Type name
        - oid (int): Type OID
        - attrs (list): Attribute descriptions
        - array_oid (int, optional): Array type OID
        - schema (str, optional): Schema name
        """
    
    def parse(self, s, curs):
        """Parse composite string."""
    
    def make(self, values):
        """Create Python object."""

def register_composite(name, conn_or_curs, globally=False, factory=None):
    """
    Register composite type.
    
    Parameters:
    - name (str): Composite type name
    - conn_or_curs: Connection or cursor
    - globally (bool): Register globally  
    - factory (callable, optional): Object factory
    """

Usage Example:

import psycopg2
from psycopg2.extras import register_composite
from collections import namedtuple

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

# Create composite type in PostgreSQL:
# CREATE TYPE address AS (street text, city text, zip text);

# Register composite type
register_composite('address', conn)

# Or with custom factory
Address = namedtuple('Address', ['street', 'city', 'zip'])
register_composite('address', conn, factory=Address)

with conn.cursor() as cur:
    # Insert composite data
    cur.execute(
        "INSERT INTO customers (name, address) VALUES (%s, %s)",
        ("John Doe", Address("123 Main St", "Anytown", "12345"))
    )
    
    # Query composite data
    cur.execute("SELECT address FROM customers WHERE name = %s", ("John Doe",))
    address = cur.fetchone()[0]
    print(f"Street: {address.street}")
    print(f"City: {address.city}")

conn.commit()
conn.close()

Types

Adapter Interface

class ISQLQuote:
    """Interface for SQL adapters."""
    
    def getquoted(self) -> bytes:
        """Return quoted SQL representation."""

Core Type Constants

# PostgreSQL type OIDs
BOOLEAN: int
BYTEA: int  
DATE: int
DECIMAL: int
FLOAT: int
INTEGER: int
INTERVAL: int
TIME: int
TIMESTAMP: int
VARCHAR: int
TEXT: int

# Array type OIDs  
BOOLEANARRAY: int
BYTESARRAY: int
DATEARRAY: int
DECIMALARRAY: int
FLOATARRAY: int
INTEGERARRAY: int
INTERVALARRAY: int
TIMEARRAY: int
STRINGARRAY: int

# JSON type OIDs
JSON_OID: int      # 114
JSONB_OID: int     # 3802
JSONARRAY_OID: int # 199
JSONBARRAY_OID: int # 3807

Range Type Interface

class Range:
    lower: Any  # Lower bound value
    upper: Any  # Upper bound value
    isempty: bool  # Empty range flag
    lower_inf: bool  # Infinite lower bound
    upper_inf: bool  # Infinite upper bound  
    lower_inc: bool  # Inclusive lower bound
    upper_inc: bool  # Inclusive upper bound
    
    def __contains__(self, x: Any) -> bool:
        """Check if value in range."""

Adaptation Registry

adapters: dict  # Global adapter registry
encodings: dict  # Encoding name mappings
string_types: tuple  # String type checking
binary_types: tuple  # Binary type checking

Install with Tessl CLI