or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

advanced-operations.mdconnections.mdcursors.mderror-handling.mdindex.mdrow-factories.mdsql-composition.mdtype-system.md
tile.json

tessl/pypi-psycopg

PostgreSQL database adapter for Python

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/psycopg@3.2.x

To install, run

npx @tessl/cli install tessl/pypi-psycopg@3.2.0

index.mddocs/

Psycopg

A modern PostgreSQL database adapter for Python that provides a comprehensive interface for connecting to and interacting with PostgreSQL databases. Psycopg 3 offers both synchronous and asynchronous operations, connection pooling, comprehensive type support, and advanced PostgreSQL features like prepared statements, server-side cursors, and COPY operations.

Package Information

  • Package Name: psycopg
  • Language: Python
  • Installation: pip install psycopg
  • Optional Extensions: pip install psycopg[binary,pool] for C speedups and connection pooling

Core Imports

import psycopg

Most common imports for database operations:

from psycopg import Connection, Cursor
from psycopg import AsyncConnection, AsyncCursor

For specific functionality:

from psycopg import sql, rows
from psycopg.types import TypeInfo, TypesRegistry
from psycopg import errors  # Exception classes
from psycopg import Pipeline, AsyncPipeline  # Pipeline operations

Basic Usage

Synchronous Connection

import psycopg

# Connect to database
with psycopg.connect("dbname=test user=postgres") as conn:
    # Execute simple query
    with conn.cursor() as cur:
        cur.execute("SELECT version()")
        version = cur.fetchone()
        print(version)
    
    # Execute parameterized query
    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO users (name, email) VALUES (%s, %s)",
            ("John Doe", "john@example.com")
        )
        conn.commit()

Asynchronous Connection

import asyncio
import psycopg

async def main():
    # Connect asynchronously
    async with await psycopg.AsyncConnection.connect("dbname=test user=postgres") as conn:
        # Execute queries asynchronously
        async with conn.cursor() as cur:
            await cur.execute("SELECT * FROM users WHERE active = %s", (True,))
            users = await cur.fetchall()
            print(users)

asyncio.run(main())

SQL Composition

from psycopg import sql

# Safe SQL composition
query = sql.SQL("INSERT INTO {table} ({fields}) VALUES ({values})").format(
    table=sql.Identifier("users"),
    fields=sql.SQL(", ").join([sql.Identifier("name"), sql.Identifier("email")]),
    values=sql.SQL(", ").join([sql.Placeholder()] * 2)
)

with conn.cursor() as cur:
    cur.execute(query, ("Jane Smith", "jane@example.com"))

Architecture

Psycopg 3 is built around several key architectural components:

  • Connection Layer: Both sync (Connection) and async (AsyncConnection) variants providing database connectivity with full transaction support
  • Cursor Hierarchy: Multiple cursor types (standard, server-side, client-side, raw) optimized for different use cases and performance requirements
  • Type System: Comprehensive PostgreSQL type adaptation supporting built-in types, arrays, JSON, custom types, and third-party integrations
  • SQL Composition: Safe query building with automatic escaping and identifier quoting
  • Pipeline Operations: Batching support for high-performance bulk operations
  • Error Handling: Complete DB-API 2.0 compliant exception hierarchy with PostgreSQL-specific diagnostics

Capabilities

Database Connections

Complete connection management for both synchronous and asynchronous operations, including connection pooling, transaction control, two-phase commit, and server configuration access.

class Connection:
    @classmethod
    def connect(cls, conninfo="", **kwargs) -> Connection: ...
    def cursor(self, *, row_factory=None) -> Cursor: ...
    def execute(self, query, params=None, *, prepare=None, binary=None) -> Cursor: ...
    def commit(self) -> None: ...
    def rollback(self) -> None: ...
    def close(self) -> None: ...

class AsyncConnection:
    @classmethod
    async def connect(cls, conninfo="", **kwargs) -> AsyncConnection: ...
    def cursor(self, *, row_factory=None) -> AsyncCursor: ...
    async def execute(self, query, params=None, *, prepare=None, binary=None) -> AsyncCursor: ...

Database Connections

Query Execution and Cursors

Comprehensive cursor functionality for query execution, result fetching, and data manipulation with support for various cursor types optimized for different performance and memory requirements.

class Cursor:
    def execute(self, query, params=None, *, prepare=None, binary=None) -> Cursor: ...
    def executemany(self, query, params_seq, *, returning=False) -> None: ...
    def fetchone(self) -> Any: ...
    def fetchmany(self, size=None) -> list: ...
    def fetchall(self) -> list: ...
    def scroll(self, value, mode="relative") -> None: ...

class ServerCursor:
    # Server-side cursor for large result sets
    
class AsyncCursor:
    async def execute(self, query, params=None, *, prepare=None, binary=None) -> AsyncCursor: ...
    async def fetchone(self) -> Any: ...

Query Execution and Cursors

Safe SQL Composition

SQL query building with automatic escaping, identifier quoting, and parameter placeholder management to prevent SQL injection while maintaining readability and flexibility.

class sql.SQL:
    def __init__(self, template: str): ...
    def format(self, *args, **kwargs) -> Composed: ...
    def join(self, seq) -> Composed: ...

class sql.Identifier:
    def __init__(self, *names): ...

class sql.Literal:
    def __init__(self, obj): ...

def sql.quote(obj, context=None) -> str: ...

Safe SQL Composition

Row Factories and Result Processing

Flexible result formatting with built-in row factories for tuples, dictionaries, named tuples, and custom objects, plus protocols for creating custom result processors.

def rows.tuple_row(cursor) -> RowMaker: ...
def rows.dict_row(cursor) -> RowMaker: ...
def rows.namedtuple_row(cursor) -> RowMaker: ...
def rows.class_row(cls) -> RowMaker: ...
def rows.scalar_row(cursor) -> RowMaker: ...

Row Factories and Result Processing

Type System and Adaptation

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

class TypeInfo:
    name: str
    oid: int
    array_oid: int
    @classmethod
    def fetch(cls, conn, name_or_oid) -> TypeInfo: ...

class TypesRegistry:
    def get_by_oid(self, oid: int) -> TypeInfo: ...
    def get_by_name(self, name: str) -> TypeInfo: ...

Type System and Adaptation

Advanced Operations

High-performance operations including COPY for bulk data transfer, pipeline operations for batching, prepared statements, and server-side cursors for memory-efficient large result set processing.

class Copy:
    def write(self, data: bytes) -> None: ...
    def write_row(self, row) -> None: ...
    def read(self) -> bytes: ...

class Pipeline:
    def sync(self) -> None: ...

Advanced Operations

Error Handling and Diagnostics

Complete DB-API 2.0 compliant exception hierarchy with PostgreSQL-specific error information, diagnostic details, and proper error classification for robust error handling.

class Error(Exception): ...
class DatabaseError(Error): ...
class DataError(DatabaseError): ...
class IntegrityError(DatabaseError): ...
class OperationalError(DatabaseError): ...
class ProgrammingError(DatabaseError): ...

Error Handling and Diagnostics

Connection Information

class ConnectionInfo:
    dsn: str
    status: int
    transaction_status: int
    server_version: int
    encoding: str
    timezone: str
    host: str
    port: int
    dbname: str
    user: str

DBAPI 2.0 Compliance

Psycopg 3 provides full DB-API 2.0 compliance with standard module-level attributes and functions:

apilevel: str = "2.0"
threadsafety: int = 2
paramstyle: str = "pyformat"

def connect(conninfo="", **kwargs) -> Connection: ...

# Type constructors
def Binary(data) -> bytes: ...
def Date(year, month, day) -> date: ...
def Time(hour, minute, second) -> time: ...
def Timestamp(year, month, day, hour, minute, second) -> datetime: ...

# Type objects for column type comparison
STRING: type
BINARY: type  
NUMBER: type
DATETIME: type
ROWID: type

# Global adapters registry access
adapters: AdaptersMap