or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

advanced-cursors.mdbatch-operations.mdconnection-pooling.mdconnections-cursors.mderror-handling.mdindex.mdreplication.mdsql-composition.mdtimezone-support.mdtype-adaptation.md
tile.json

tessl/pypi-psycopg2

Python-PostgreSQL Database Adapter

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/psycopg2@2.9.x

To install, run

npx @tessl/cli install tessl/pypi-psycopg2@2.9.0

index.mddocs/

psycopg2

psycopg2 is the most popular PostgreSQL database adapter for the Python programming language. It provides complete implementation of the Python DB API 2.0 specification with thread safety for heavily multi-threaded applications. The library is primarily implemented in C as a libpq wrapper, delivering both efficiency and security with support for client-side and server-side cursors, asynchronous communication and notifications, COPY TO/COPY FROM operations, and comprehensive type adaptation between Python and PostgreSQL data types.

Package Information

  • Package Name: psycopg2
  • Language: Python
  • Installation: pip install psycopg2 or pip install psycopg2-binary

Note: psycopg2-binary provides pre-compiled wheels for easier installation, while psycopg2 requires compilation and PostgreSQL development headers.

Core Imports

import psycopg2

For specific functionality:

from psycopg2 import connect, Error
from psycopg2.extensions import connection, cursor, adapt, register_adapter
from psycopg2.extras import DictCursor, RealDictCursor, execute_batch, execute_values
from psycopg2.sql import SQL, Identifier, Literal
from psycopg2.pool import SimpleConnectionPool, ThreadedConnectionPool
from psycopg2.tz import FixedOffsetTimezone, LocalTimezone

Basic Usage

import psycopg2
from psycopg2.extras import RealDictCursor

# Connect to PostgreSQL database
conn = psycopg2.connect(
    host="localhost",
    database="mydb",
    user="myuser",
    password="mypassword"
)

# Create cursor and execute query
with conn.cursor(cursor_factory=RealDictCursor) as cur:
    cur.execute("SELECT * FROM users WHERE age > %s", (25,))
    users = cur.fetchall()
    
    for user in users:
        print(f"{user['name']}: {user['email']}")

# Insert data
with conn.cursor() as cur:
    cur.execute(
        "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
        ("John Doe", "john@example.com", 30)
    )
    conn.commit()

conn.close()

Architecture

psycopg2 follows the Python DB API 2.0 specification with these core components:

  • Connection Objects: Database connections with transaction management and configuration
  • Cursor Objects: Execute queries and fetch results with various cursor types for different use cases
  • Type System: Comprehensive adaptation between Python and PostgreSQL types including arrays, JSON, ranges, and custom types
  • Extensions: Advanced features like async operations, connection pooling, and specialized data types
  • Error Handling: Complete PostgreSQL error code mapping to Python exceptions

The library supports both synchronous and asynchronous operations, providing flexibility for different application architectures from simple scripts to high-performance web applications.

Capabilities

Database Connections and Cursors

Core database connectivity with connection management, cursor operations, transaction handling, and various cursor types for different result formats.

def connect(dsn=None, connection_factory=None, cursor_factory=None, **kwargs):
    """Create a new database connection."""

class connection:
    def cursor(self, name=None, cursor_factory=None):
        """Create a new cursor."""
    def commit(self):
        """Commit current transaction."""
    def rollback(self):
        """Rollback current transaction."""
    def close(self):
        """Close the connection."""

class cursor:
    def execute(self, query, vars=None):
        """Execute a database operation."""
    def fetchone(self):
        """Fetch the next row of query result."""
    def fetchmany(self, size=None):
        """Fetch multiple rows."""
    def fetchall(self):
        """Fetch all remaining rows."""

Database Connections and Cursors

Advanced Cursor Types

Specialized cursor classes that return results as dictionaries, named tuples, or provide logging capabilities for development and debugging.

class DictCursor(cursor):
    """Cursor returning dict-like rows."""

class RealDictCursor(cursor):
    """Cursor with real dict rows."""

class NamedTupleCursor(cursor):
    """Cursor returning named tuples."""

class LoggingCursor(cursor):
    """Cursor that logs executed queries."""

Advanced Cursor Types

SQL Composition

Safe SQL query construction with automatic quoting, parameter placeholders, and composable SQL elements to prevent SQL injection vulnerabilities.

class SQL:
    """Raw SQL snippet."""
    def __init__(self, string): ...
    def format(self, *args, **kwargs): ...

class Identifier:
    """SQL identifier (quoted)."""
    def __init__(self, *strings): ...

class Literal:
    """SQL literal value."""

class Placeholder:
    """Parameter placeholder."""

SQL Composition

Type Adaptation and Casting

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

def adapt(obj):
    """Adapt Python object to SQL."""

def register_adapter(type, adapter):
    """Register object adapter."""

def new_type(oids, name, castfunc):
    """Create new typecaster."""

def register_type(obj, scope=None):
    """Register typecaster."""

class Binary:
    """Binary data adapter."""

class Json:
    """JSON adapter class."""

Type Adaptation and Casting

Connection Pooling

Thread-safe and non-thread-safe connection pools for managing database connections efficiently in multi-threaded applications.

class SimpleConnectionPool:
    """Non-threadsafe connection pool."""
    def __init__(self, minconn, maxconn, *args, **kwargs): ...
    def getconn(self, key=None): ...
    def putconn(self, conn, key=None, close=False): ...

class ThreadedConnectionPool:
    """Thread-safe connection pool."""
    def __init__(self, minconn, maxconn, *args, **kwargs): ...
    def getconn(self, key=None): ...
    def putconn(self, conn=None, key=None, close=False): ...

Connection Pooling

PostgreSQL Replication

Logical and physical replication support for PostgreSQL streaming replication, including replication slot management and message handling.

class LogicalReplicationConnection:
    """Logical replication connection."""

class PhysicalReplicationConnection:
    """Physical replication connection."""

class ReplicationCursor:
    """Cursor for replication connections."""
    def create_replication_slot(self, slot_name, slot_type=None, output_plugin=None): ...
    def start_replication(self, slot_name=None, **kwargs): ...

PostgreSQL Replication

Error Handling and Diagnostics

Complete PostgreSQL error code mapping to Python exceptions with detailed error information and diagnostic capabilities.

class Error(Exception):
    """Base exception class."""

class DatabaseError(Error):
    """Database engine errors."""

class IntegrityError(DatabaseError):
    """Database integrity violations."""

class ProgrammingError(DatabaseError):
    """SQL programming errors."""

def lookup(code):
    """Look up exception class by error code."""

Error Handling and Diagnostics

Batch Operations and Utilities

Efficient batch execution functions and utility operations for improved performance with multiple queries and specialized database operations.

def execute_batch(cur, sql, argslist, page_size=100):
    """Execute SQL with multiple parameter sets efficiently."""

def execute_values(cur, sql, argslist, template=None, page_size=100, fetch=False):
    """Execute INSERT with VALUES clause."""

def wait_select(conn):
    """Wait callback for select-based waiting."""

def make_dsn(dsn=None, **kwargs):
    """Build connection string from parameters."""

def parse_dsn(dsn):
    """Parse connection string into components."""

def quote_ident(name, scope=None):
    """Quote SQL identifier."""

Batch Operations and Utilities

Timezone Support

Timezone handling utilities for PostgreSQL timestamp types with timezone information, including fixed offset and local timezone support.

class FixedOffsetTimezone(datetime.tzinfo):
    """Fixed UTC offset timezone."""
    def __init__(self, offset=None, name=None): ...

class LocalTimezone(datetime.tzinfo):
    """Platform's local timezone."""

LOCAL: LocalTimezone  # Local timezone instance
ZERO: timedelta  # Zero timedelta

Timezone Support

Types

Core Database API Types

# DB API 2.0 constants
apilevel: str  # '2.0'
threadsafety: int  # 2
paramstyle: str  # 'pyformat'

# Version information
__version__: str
__libpq_version__: int

# Type constants
BINARY: type
NUMBER: type  
STRING: type
DATETIME: type
ROWID: type

Connection and Transaction Constants

# Isolation levels
ISOLATION_LEVEL_AUTOCOMMIT: int  # 0
ISOLATION_LEVEL_READ_UNCOMMITTED: int  # 4
ISOLATION_LEVEL_READ_COMMITTED: int  # 1
ISOLATION_LEVEL_REPEATABLE_READ: int  # 2
ISOLATION_LEVEL_SERIALIZABLE: int  # 3

# Connection status
STATUS_READY: int  # 1
STATUS_BEGIN: int  # 2
STATUS_IN_TRANSACTION: int  # STATUS_BEGIN

# Transaction status  
TRANSACTION_STATUS_IDLE: int  # 0
TRANSACTION_STATUS_ACTIVE: int  # 1
TRANSACTION_STATUS_INTRANS: int  # 2
TRANSACTION_STATUS_INERROR: int  # 3
TRANSACTION_STATUS_UNKNOWN: int  # 4

Polling Constants

# Asynchronous connection polling
POLL_OK: int  # 0
POLL_READ: int  # 1
POLL_WRITE: int  # 2
POLL_ERROR: int  # 3

Replication Constants

# Replication types
REPLICATION_PHYSICAL: int  # Physical replication
REPLICATION_LOGICAL: int  # Logical replication

Data Type Constructors

def Binary(obj) -> bytes:
    """Create binary data object."""

def Date(year: int, month: int, day: int):
    """Create date object."""

def Time(hour: int, minute: int, second: int):
    """Create time object."""

def Timestamp(year: int, month: int, day: int, hour: int, minute: int, second: int):
    """Create timestamp object."""

def DateFromTicks(ticks: float):
    """Create date from timestamp."""

def TimeFromTicks(ticks: float):
    """Create time from timestamp."""

def TimestampFromTicks(ticks: float):
    """Create timestamp from timestamp."""