tessl/pypi-psycopg2-binary

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

—

Pending

Overview

Eval results

Files

Type Adaptation and PostgreSQL Types

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

Comprehensive type system for converting between Python objects and PostgreSQL data types. psycopg2 provides automatic adaptation for common types and extensible mechanisms for custom types including arrays, JSON/JSONB, ranges, IP addresses, and user-defined types.

Capabilities

Core Type Adaptation

Fundamental type adaptation system allowing registration of adapters and type casters for bidirectional conversion between Python and PostgreSQL types.

def adapt(obj):
    """
    Adapt Python object for PostgreSQL.
    
    Parameters:
    - obj: Python object to adapt
    
    Returns:
    ISQLQuote: Adapted object ready for SQL inclusion
    """

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

def new_type(oids, name, castfunc):
    """
    Create new type caster for PostgreSQL type.
    
    Parameters:
    - oids (tuple): PostgreSQL type OIDs
    - name (str): Type name
    - castfunc (callable): Cast function
    
    Returns:
    Type caster object
    """

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

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

JSON and JSONB Support

Native support for PostgreSQL JSON and JSONB data types with automatic serialization/deserialization and custom encoder support.

class Json:
    """JSON adapter for PostgreSQL json/jsonb types."""
    
    def __init__(self, adapted, dumps=None):
        """
        Initialize JSON adapter.
        
        Parameters:
        - adapted: Python object to adapt
        - dumps (callable, optional): Custom JSON encoder
        """
    
    def dumps(self, obj):
        """
        Serialize object to JSON.
        
        Parameters:
        - obj: Object to serialize
        
        Returns:
        str: JSON string
        """

def register_json(conn_or_curs=None, globally=False, loads=None, oid=None, array_oid=None, name='json'):
    """
    Register JSON type casters.
    
    Parameters:
    - conn_or_curs (connection/cursor, optional): Registration scope
    - globally (bool): Register globally
    - loads (callable, optional): Custom JSON decoder
    - oid (int, optional): JSON type OID
    - array_oid (int, optional): JSON array type OID
    - name (str): Type name to register
    
    Returns:
    tuple: (JSON_caster, JSONARRAY_caster)
    """

def register_default_json(conn_or_curs=None, globally=False, loads=None):
    """
    Register default JSON type casters for PostgreSQL 9.2+.
    
    Parameters:
    - conn_or_curs (connection/cursor, optional): Registration scope
    - globally (bool): Register globally
    - loads (callable, optional): Custom JSON decoder
    
    Returns:
    tuple: (JSON_caster, JSONARRAY_caster)
    """

def register_default_jsonb(conn_or_curs=None, globally=False, loads=None):
    """
    Register default JSONB type casters for PostgreSQL 9.4+.
    
    Parameters:
    - conn_or_curs (connection/cursor, optional): Registration scope  
    - globally (bool): Register globally
    - loads (callable, optional): Custom JSON decoder
    
    Returns:
    tuple: (JSONB_caster, JSONBARRAY_caster)
    """

Usage examples:

# Basic JSON usage
from psycopg2.extras import Json
data = {'name': 'Alice', 'age': 30}
cur.execute("INSERT INTO users (data) VALUES (%s)", (Json(data),))

# Retrieve JSON data
cur.execute("SELECT data FROM users WHERE id = %s", (1,))
user_data = cur.fetchone()[0]  # Automatically decoded to Python dict

# Custom JSON encoder/decoder
import json
def custom_encoder(obj):
    return json.dumps(obj, indent=2)

def custom_decoder(s):
    return json.loads(s)

# Register with custom functions
register_default_json(loads=custom_decoder)
cur.execute("INSERT INTO users (data) VALUES (%s)", (Json(data, dumps=custom_encoder),))

Range Types

Support for PostgreSQL range types including numeric ranges, date ranges, and timestamp ranges.

class Range:
    """Base class for PostgreSQL range types."""
    
    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 flags ('[)', '(]', '()', '[]')
        - empty (bool): Create empty range
        """
    
    @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 included."""
        
    @property
    def upper_inc(self):
        """True if upper bound is included."""
    
    def __contains__(self, x):
        """Check if value is in range."""

class NumericRange(Range):
    """Range for numeric types (int4range, int8range, numrange)."""

class DateRange(Range):
    """Range for date type (daterange)."""

class DateTimeRange(Range):
    """Range for timestamp type (tsrange)."""

class DateTimeTZRange(Range):
    """Range for timestamp with timezone type (tstzrange)."""

def register_range(pgrange, pyrange, conn_or_curs=None, globally=False):
    """
    Register range type adapter and caster.
    
    Parameters:
    - pgrange (str): PostgreSQL range type name
    - pyrange (type): Python range class
    - conn_or_curs (connection/cursor, optional): Registration scope
    - globally (bool): Register globally
    """
    
class RangeAdapter:
    """Adapter for range types."""
    
    def __init__(self, wrapped):
        """Initialize range adapter."""

class RangeCaster:
    """Type caster for range types."""
    
    def __init__(self, pgrange, pyrange, parse):
        """Initialize range caster."""

Usage examples:

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

# Numeric ranges
num_range = NumericRange(1, 10)  # [1,10)
num_range_inclusive = NumericRange(1, 10, '[]')  # [1,10]
empty_range = NumericRange(empty=True)

# Date ranges  
date_range = DateRange(date(2023, 1, 1), date(2023, 12, 31))

# Check containment
if 5 in num_range:
    print("5 is in range")

# Store ranges
cur.execute("INSERT INTO events (date_range) VALUES (%s)", (date_range,))

# Retrieve ranges
cur.execute("SELECT date_range FROM events WHERE id = %s", (1,))
retrieved_range = cur.fetchone()[0]
print(f"Range: {retrieved_range.lower} to {retrieved_range.upper}")

IP Address Support

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

def register_ipaddress(conn_or_curs=None):
    """
    Register IP address type support for PostgreSQL network types.
    
    Parameters:
    - conn_or_curs (connection/cursor, optional): Registration scope. If None, register globally.
    
    Note:
    - inet values convert to IPv4Interface/IPv6Interface objects
    - cidr values convert to IPv4Network/IPv6Network objects
    """

def cast_interface(s, cur=None):
    """Cast PostgreSQL inet value to ipaddress interface object."""

def cast_network(s, cur=None):
    """Cast PostgreSQL cidr value to ipaddress network object."""

def adapt_ipaddress(obj):
    """Adapt ipaddress objects for PostgreSQL."""

Usage examples:

from psycopg2._ipaddress import register_ipaddress
import ipaddress

# Register IP address support
register_ipaddress()

# Use Python ipaddress objects
interface = ipaddress.ip_interface('192.168.1.1/24')  # inet type
network = ipaddress.ip_network('192.168.1.0/24')     # cidr type

cur.execute("INSERT INTO hosts (interface, network) VALUES (%s, %s)", (interface, network))

# Retrieve as ipaddress objects
cur.execute("SELECT interface, network FROM hosts WHERE id = %s", (1,)) 
retrieved_interface, retrieved_network = cur.fetchone()
print(f"Interface: {retrieved_interface}")  # IPv4Interface object
print(f"Network: {retrieved_network}")      # IPv4Network object

# Works with both IPv4 and IPv6
ipv6_interface = ipaddress.ip_interface('::1/128')
cur.execute("INSERT INTO hosts (interface) VALUES (%s)", (ipv6_interface,))

Array Types

Automatic support for PostgreSQL array types with bidirectional conversion to Python lists.

# Arrays are automatically handled
int_array = [1, 2, 3, 4, 5]
text_array = ['hello', 'world']

cur.execute("INSERT INTO data (integers, texts) VALUES (%s, %s)", 
           (int_array, text_array))

# Retrieve arrays
cur.execute("SELECT integers, texts FROM data WHERE id = %s", (1,))
ints, texts = cur.fetchone()
print(f"Integers: {ints}, Texts: {texts}")  # Python lists

# Multidimensional arrays
matrix = [[1, 2], [3, 4]]
cur.execute("INSERT INTO matrices (data) VALUES (%s)", (matrix,))

UUID Support

Support for PostgreSQL UUID type using Python's uuid module.

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

Usage examples:

from psycopg2.extras import register_uuid
import uuid

# Register UUID support
register_uuid()

# Use Python UUID objects
user_id = uuid.uuid4()
cur.execute("INSERT INTO users (id, name) VALUES (%s, %s)", (user_id, 'Alice'))

# Retrieve as UUID objects
cur.execute("SELECT id FROM users WHERE name = %s", ('Alice',))
retrieved_id = cur.fetchone()[0]
print(f"User ID: {retrieved_id}")  # UUID object

Timezone Support

Timezone utilities for working with PostgreSQL timestamp types and time zone conversions.

class FixedOffsetTimezone(datetime.tzinfo):
    """
    Fixed offset timezone implementation.
    
    Parameters:
    - offset (timedelta/int, optional): UTC offset as timedelta or minutes
    - name (str, optional): Timezone name
    """
    
    def __init__(self, offset=None, name=None): ...
    def utcoffset(self, dt): ...
    def tzname(self, dt): ...
    def dst(self, dt): ...

class LocalTimezone(datetime.tzinfo):
    """Platform's local timezone implementation."""
    
    def utcoffset(self, dt): ...
    def dst(self, dt): ...
    def tzname(self, dt): ...

# Constants
LOCAL: LocalTimezone  # Local timezone instance
ZERO: datetime.timedelta  # Zero timedelta constant

Usage examples:

from psycopg2.tz import FixedOffsetTimezone, LocalTimezone, LOCAL
import datetime

# Create fixed offset timezone (UTC+5:30)
ist = FixedOffsetTimezone(offset=330)  # 330 minutes
# or using timedelta
ist = FixedOffsetTimezone(offset=datetime.timedelta(hours=5, minutes=30))

# Create datetime with timezone
dt = datetime.datetime(2023, 1, 1, 12, 0, 0, tzinfo=ist)

# Use local timezone
local_dt = datetime.datetime(2023, 1, 1, 12, 0, 0, tzinfo=LOCAL)

# Store timezone-aware datetime
cur.execute("INSERT INTO events (event_time) VALUES (%s)", (dt,))

# Retrieve with timezone information preserved
cur.execute("SELECT event_time FROM events")
retrieved_dt = cur.fetchone()[0]
print(f"Event time: {retrieved_dt}")  # Includes timezone info

HStore Support

Support for PostgreSQL hstore extension providing key-value storage.

def register_hstore(conn_or_curs, globally=False, unicode=False, oid=None, array_oid=None):
    """
    Register hstore type support.
    
    Parameters:
    - conn_or_curs (connection/cursor): Connection to query hstore OIDs
    - globally (bool): Register globally
    - unicode (bool): Return unicode strings
    - oid (int, optional): hstore type OID
    - array_oid (int, optional): hstore array type OID
    """

class HstoreAdapter:
    """Adapter for hstore type."""
    
    def __init__(self, wrapped):
        """Initialize hstore adapter."""

Usage examples:

from psycopg2.extras import register_hstore

# Register hstore (requires hstore extension in database)
register_hstore(conn)

# Use Python dictionaries
data = {'key1': 'value1', 'key2': 'value2'}
cur.execute("INSERT INTO products (attributes) VALUES (%s)", (data,))

# Retrieve as dictionaries
cur.execute("SELECT attributes FROM products WHERE id = %s", (1,))
attrs = cur.fetchone()[0]
print(f"Attributes: {attrs}")  # Python dict

Composite Types

Support for PostgreSQL composite types (user-defined row types).

def register_composite(name, conn_or_curs, globally=False, factory=None):
    """
    Register composite type support.
    
    Parameters:
    - name (str): Composite type name
    - conn_or_curs (connection/cursor): Connection to query type info
    - globally (bool): Register globally
    - factory (callable, optional): Factory function for composite objects
    
    Returns:
    Type caster for composite type
    """

class CompositeCaster:
    """Type caster for composite types."""
    
    def __init__(self, name, oid, attrs, array_oid=None, schema=None):
        """Initialize composite caster."""

Usage examples:

from psycopg2.extras import register_composite
from collections import namedtuple

# Create composite type in database
cur.execute("""
    CREATE TYPE person AS (
        name text,
        age integer,
        email text
    )
""")

# Register composite type
Person = register_composite('person', cur)

# Use composite type
person = Person('Alice', 30, 'alice@example.com')
cur.execute("INSERT INTO people (data) VALUES (%s)", (person,))

# Retrieve composite objects
cur.execute("SELECT data FROM people WHERE id = %s", (1,))
retrieved_person = cur.fetchone()[0]
print(f"Name: {retrieved_person.name}, Age: {retrieved_person.age}")

Types

Adapter Interface

class ISQLQuote:
    """Interface for SQL quotable objects."""
    
    def getquoted(self):
        """Return quoted representation as bytes."""
    
    def prepare(self, conn):
        """Prepare adapter for specific connection."""

Built-in Adapters

class AsIs:
    """Pass-through adapter (no quoting)."""
    
class Binary:
    """Binary data adapter."""
    
class Boolean:
    """Boolean adapter."""
    
class Float:
    """Float adapter."""
    
class Int:
    """Integer adapter."""
    
class QuotedString:
    """String adapter with proper quoting."""

Type Constants

# PostgreSQL type OIDs
BOOLEAN: int = 16
BYTEA: int = 17
INT8: int = 20
INT2: int = 21
INT4: int = 23
TEXT: int = 25
FLOAT4: int = 700
FLOAT8: int = 701
TIMESTAMP: int = 1114
TIMESTAMPTZ: int = 1184
DATE: int = 1082
TIME: int = 1083

# Array type OIDs  
BOOLEANARRAY: int = 1000
INTEGERARRAY: int = 1007
FLOATARRAY: int = 1021
STRINGARRAY: int = 1009

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

Install with Tessl CLI