tessl/pypi-py-postgresql

PostgreSQL driver and tools library providing PG-API and DB-API 2.0 interfaces for Python.

—

Pending

Overview

Eval results

Files

Type System

Name: tessl/pypi-py-postgresql
Author: tessl

Comprehensive PostgreSQL type support including primitive types, arrays, composite types, and custom type conversion with automatic serialization and deserialization.

Capabilities

PostgreSQL Type Constants

PostgreSQL type OID constants for all standard data types, enabling type identification and custom type handling.

# Numeric types
BOOLOID: int        # 16 - Boolean
INT2OID: int        # 21 - Smallint (2 bytes)
INT4OID: int        # 23 - Integer (4 bytes)  
INT8OID: int        # 20 - Bigint (8 bytes)
FLOAT4OID: int      # 700 - Real (4 bytes)
FLOAT8OID: int      # 701 - Double precision (8 bytes)
NUMERICOID: int     # 1700 - Numeric/Decimal

# String types
TEXTOID: int        # 25 - Text
VARCHAROID: int     # 1043 - Varchar
CHAROID: int        # 1042 - Char
NAMEOID: int        # 19 - Name
BYTEAOID: int       # 17 - Bytea (binary data)

# Date/time types
DATEOID: int        # 1082 - Date
TIMEOID: int        # 1083 - Time
TIMESTAMPOID: int   # 1114 - Timestamp
TIMESTAMPTZOID: int # 1184 - Timestamp with timezone
INTERVALOID: int    # 1186 - Interval

# JSON types
JSONOID: int        # 114 - JSON
JSONBOID: int       # 3802 - JSONB

# Other common types
UUIDOID: int        # 2950 - UUID
XMLOID: int         # 142 - XML
INETOID: int        # 869 - inet (IP address)
CIDROID: int        # 650 - cidr (IP network)
MACADDROID: int     # 829 - macaddr (MAC address)

# Array type OIDs (add 1000 to base type for array)
TEXTARRAYOID: int   # 1009 - Text array
INT4ARRAYOID: int   # 1007 - Integer array

Type Mapping Utilities

Dictionaries for converting between PostgreSQL type OIDs, names, and SQL type names.

oid_to_name: dict
    """
    Maps PostgreSQL type OIDs to type names.
    
    Example: {23: 'int4', 25: 'text', 1700: 'numeric'}
    """

oid_to_sql_name: dict
    """
    Maps PostgreSQL type OIDs to SQL standard type names.
    
    Example: {23: 'INTEGER', 25: 'TEXT', 1700: 'NUMERIC'}
    """

name_to_oid: dict
    """
    Maps PostgreSQL type names to type OIDs.
    
    Example: {'int4': 23, 'text': 25, 'numeric': 1700}
    """

Array Type

PostgreSQL array type with multi-dimensional support and automatic element type conversion.

class Array:
    """
    PostgreSQL array type supporting multi-dimensional arrays with automatic
    element type conversion and nested array operations.
    """
    
    def __init__(elements, element_type):
        """
        Create PostgreSQL array.
        
        Parameters:
        - elements (list): Array elements or nested lists for multi-dimensional
        - element_type (int): PostgreSQL type OID for array elements
        """
    
    def __getitem__(index):
        """
        Get array element by index.
        
        Parameters:
        - index (int or slice): Element index or slice
        
        Returns:
        Element value with automatic type conversion
        """
    
    def __setitem__(index, value):
        """
        Set array element value.
        
        Parameters:
        - index (int): Element index
        - value: New element value
        """
    
    def __len__():
        """
        Get array length.
        
        Returns:
        int: Number of elements in array
        """
    
    def __iter__():
        """Iterate over array elements."""
    
    def append(value):
        """
        Append element to array.
        
        Parameters:
        - value: Element to append
        """
    
    def extend(values):
        """
        Extend array with multiple elements.
        
        Parameters:
        - values (iterable): Elements to add
        """
    
    @property
    def dimensions():
        """
        Get array dimensions.
        
        Returns:
        tuple: Dimension sizes for multi-dimensional arrays
        """
    
    @property
    def element_type():
        """
        Get element type OID.
        
        Returns:
        int: PostgreSQL type OID for array elements
        """

Row Type

Named tuple-like interface for query result rows with both positional and named access.

class Row:
    """
    Result row providing named and positional access to column values
    with automatic type conversion.
    """
    
    def __getitem__(key):
        """
        Get column value by index or name.
        
        Parameters:
        - key (int or str): Column index (0-based) or column name
        
        Returns:
        Column value with automatic PostgreSQL type conversion
        """
    
    def __len__():
        """
        Get number of columns.
        
        Returns:
        int: Number of columns in row
        """
    
    def __iter__():
        """Iterate over column values."""
    
    def keys():
        """
        Get column names.
        
        Returns:
        list[str]: Column names in order
        """
    
    def values():
        """
        Get column values.
        
        Returns:
        list: Column values in order
        """
    
    def items():
        """
        Get column name-value pairs.
        
        Returns:
        list[tuple]: (name, value) pairs
        """
    
    def get(key, default=None):
        """
        Get column value with default.
        
        Parameters:
        - key (str): Column name
        - default: Default value if column not found
        
        Returns:
        Column value or default
        """

Type I/O Interface

Type conversion interface for custom PostgreSQL type handling.

class TypeIO:
    """
    Type input/output interface for PostgreSQL type conversion.
    """
    
    def encode(value, format):
        """
        Encode Python value to PostgreSQL format.
        
        Parameters:
        - value: Python value to encode
        - format (str): Output format ('text' or 'binary')
        
        Returns:
        bytes: Encoded value for PostgreSQL
        """
    
    def decode(data, format):
        """
        Decode PostgreSQL data to Python value.
        
        Parameters:
        - data (bytes): PostgreSQL data
        - format (str): Input format ('text' or 'binary')
        
        Returns:
        Python value with appropriate type
        """

Usage Examples

Working with Arrays

import postgresql
import postgresql.types as pg_types

db = postgresql.open('pq://user:pass@localhost/mydb')

# Create table with array columns
db.execute("""
    CREATE TABLE IF NOT EXISTS test_arrays (
        id SERIAL PRIMARY KEY,
        numbers INTEGER[],
        texts TEXT[],
        matrix INTEGER[][]
    )
""")

# Insert arrays using Python lists
insert_stmt = db.prepare("""
    INSERT INTO test_arrays (numbers, texts, matrix) 
    VALUES ($1, $2, $3)
""")

# PostgreSQL automatically converts Python lists to arrays
insert_stmt([1, 2, 3, 4, 5], ['hello', 'world'], [[1, 2], [3, 4]])

# Query arrays
get_arrays = db.prepare("SELECT numbers, texts, matrix FROM test_arrays WHERE id = $1")
result = get_arrays.first(1)

numbers = result['numbers']  # Returns Python list
texts = result['texts']      # Returns Python list  
matrix = result['matrix']    # Returns nested Python list

print(f"Numbers: {numbers}")        # [1, 2, 3, 4, 5]
print(f"Texts: {texts}")            # ['hello', 'world']
print(f"Matrix: {matrix}")          # [[1, 2], [3, 4]]

# Work with array elements
print(f"First number: {numbers[0]}")
print(f"Matrix element [1][0]: {matrix[1][0]}")

Type Identification and Conversion

import postgresql
import postgresql.types as pg_types

db = postgresql.open('pq://user:pass@localhost/mydb')

# Prepare statement and check parameter types
stmt = db.prepare("SELECT $1::text, $2::integer, $3::numeric, $4::json")

print("Parameter types:")
for i, type_oid in enumerate(stmt.parameter_types):
    type_name = pg_types.oid_to_name.get(type_oid, f"unknown({type_oid})")
    sql_name = pg_types.oid_to_sql_name.get(type_oid, f"unknown({type_oid})")
    print(f"  ${i+1}: {type_name} (SQL: {sql_name})")

# Execute with various types
result = stmt.first("hello", 42, 123.45, {"key": "value"})

print("Result types:")
for i, type_oid in enumerate(stmt.result_types):
    type_name = pg_types.oid_to_name.get(type_oid, f"unknown({type_oid})")
    print(f"  Column {i}: {type_name}")

print(f"Results: {result}")

Custom Type Handling

import postgresql
import postgresql.types as pg_types
import json
from decimal import Decimal

db = postgresql.open('pq://user:pass@localhost/mydb')

# Create table with various types
db.execute("""
    CREATE TABLE IF NOT EXISTS type_examples (
        id SERIAL PRIMARY KEY,
        json_data JSONB,
        money_amount NUMERIC(10,2),
        uuid_value UUID,
        ip_address INET,
        created_at TIMESTAMP WITH TIME ZONE
    )
""")

# Insert with automatic type conversion
insert_stmt = db.prepare("""
    INSERT INTO type_examples (json_data, money_amount, uuid_value, ip_address, created_at)
    VALUES ($1, $2, $3, $4, $5)
""")

import uuid
from datetime import datetime

insert_stmt(
    {"name": "John", "age": 30, "active": True},  # Dict -> JSONB
    Decimal('1234.56'),                           # Decimal -> NUMERIC
    uuid.uuid4(),                                 # UUID -> UUID
    '192.168.1.1',                               # str -> INET
    datetime.now()                               # datetime -> TIMESTAMPTZ
)

# Query with automatic conversion back to Python types
select_stmt = db.prepare("SELECT * FROM type_examples WHERE id = $1")
result = select_stmt.first(1)

print(f"JSON data: {result['json_data']}")      # dict
print(f"Money: {result['money_amount']}")       # Decimal
print(f"UUID: {result['uuid_value']}")          # UUID
print(f"IP: {result['ip_address']}")            # str
print(f"Created: {result['created_at']}")       # datetime

Working with Composite Types

import postgresql

db = postgresql.open('pq://user:pass@localhost/mydb')

# Create composite type
db.execute("""
    CREATE TYPE IF NOT EXISTS address AS (
        street TEXT,
        city TEXT,
        state TEXT,
        zip_code TEXT
    )
""")

# Create table using composite type
db.execute("""
    CREATE TABLE IF NOT EXISTS customers (
        id SERIAL PRIMARY KEY,
        name TEXT,
        address address
    )
""")

# Insert using tuple or dict for composite type
insert_stmt = db.prepare("""
    INSERT INTO customers (name, address) 
    VALUES ($1, $2)
""")

# Can use tuple (positional) or dict (named) for composite types
insert_stmt("John Doe", ("123 Main St", "Anytown", "CA", "12345"))

# Query composite type
select_stmt = db.prepare("SELECT id, name, address FROM customers WHERE id = $1")
result = select_stmt.first(1)

address = result['address']  # Returns Row object for composite type
print(f"Customer: {result['name']}")
print(f"Address: {address['street']}, {address['city']}, {address['state']} {address['zip_code']}")

# Access composite type components
print(f"Street: {address['street']}")
print(f"City: {address['city']}")

Type Introspection

import postgresql
import postgresql.types as pg_types

db = postgresql.open('pq://user:pass@localhost/mydb')

# Get information about all types in database
type_query = db.prepare("""
    SELECT oid, typname, typtype, typlen
    FROM pg_type 
    WHERE typtype IN ('b', 'c', 'e')  -- base, composite, enum types
    ORDER BY typname
""")

print("Database types:")
for row in type_query():
    type_oid = row['oid']
    type_name = row['typname']
    type_type = row['typtype']
    type_len = row['typlen']
    
    # Check if we have this type in our mappings
    known_name = pg_types.oid_to_name.get(type_oid, "unknown")
    sql_name = pg_types.oid_to_sql_name.get(type_oid, "")
    
    type_category = {
        'b': 'base',
        'c': 'composite', 
        'e': 'enum'
    }.get(type_type, 'other')
    
    print(f"  {type_name} (OID {type_oid}): {type_category}, known as '{known_name}', SQL: '{sql_name}'")

# Check array types
print("\nArray type examples:")
array_types = [
    (pg_types.TEXTARRAYOID, "text[]"),
    (pg_types.INT4ARRAYOID, "integer[]"),
    (pg_types.JSONBOID + 1000, "jsonb[]")  # Array OID = base OID + 1000
]

for array_oid, description in array_types:
    if array_oid in pg_types.oid_to_name:
        print(f"  {description}: OID {array_oid}")

Install with Tessl CLI