tessl/pypi-pg8000

Pure-Python PostgreSQL driver providing DB-API 2.0 compliant database connectivity with native pg8000 API and comprehensive PostgreSQL data type support.

—

Pending

Overview

Eval results

Files

Connection Management

Name: tessl/pypi-pg8000
Author: tessl

Primary functions and patterns for establishing and managing database connections to PostgreSQL servers using pg8000.

Capabilities

Primary Connection Function

Creates a connection to a PostgreSQL database server using the legacy DB-API 2.0 interface.

def connect(
    user: str,
    host: str = "localhost",
    database: str = None, 
    port: int = 5432,
    password: str = None,
    source_address: tuple = None,
    unix_sock: str = None,
    ssl_context = None,
    timeout: float = None,
    tcp_keepalive: bool = True,
    application_name: str = None,
    replication: str = None,
    startup_params: dict = None,
    sock = None
) -> Connection:
    """
    Create a connection to a PostgreSQL database.
    
    Parameters:
    - user: Username for database authentication
    - host: Database server hostname or IP address
    - database: Database name to connect to (defaults to username)
    - port: Database server port number
    - password: Password for database authentication
    - source_address: (host, port) tuple for local socket binding
    - unix_sock: Path to Unix domain socket for local connections
    - ssl_context: SSL context for encrypted connections
    - timeout: Socket timeout in seconds
    - tcp_keepalive: Enable TCP keepalive for connection health
    - application_name: Application name for connection identification
    - replication: Replication mode ('database' or 'physical')
    - startup_params: Additional connection parameters dict
    - sock: Existing socket connection for database communication
    
    Returns:
    Connection object for database operations
    
    Raises:
    InterfaceError: Connection parameter or interface errors
    OperationalError: Database connection errors
    """

Module Constants

Module-level constants defining pg8000's DB-API 2.0 compliance characteristics.

apilevel: str = "2.0"
"""DBAPI level supported by pg8000."""

threadsafety: int = 1  
"""Thread safety level - module sharing only, connections not thread-safe."""

paramstyle: str = "format"
"""Parameter style used for SQL queries."""

Connection Examples

Basic Connection

import pg8000

# Minimal connection
conn = pg8000.connect(
    user="myuser",
    password="mypassword",
    database="mydb"
)

# Use connection
cursor = conn.cursor()
cursor.execute("SELECT current_user")
print(cursor.fetchone())

# Clean up
cursor.close() 
conn.close()

Advanced Connection Options

import pg8000
import ssl

# Connection with SSL and advanced options
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

conn = pg8000.connect(
    user="myuser",
    password="mypassword", 
    host="db.example.com",
    port=5432,
    database="production_db",
    ssl_context=ssl_context,
    timeout=30.0,
    tcp_keepalive=True,
    application_name="MyApp v1.0",
    startup_params={
        "search_path": "public,audit",
        "timezone": "UTC"
    }
)

# Connection is ready for use
cursor = conn.cursor()
cursor.execute("SELECT version()")
print(cursor.fetchone())

cursor.close()
conn.close()

Unix Socket Connection

import pg8000

# Connect using Unix domain socket
conn = pg8000.connect(
    user="myuser",
    unix_sock="/var/run/postgresql/.s.PGSQL.5432",
    database="mydb"
)

# Use connection normally
cursor = conn.cursor()
cursor.execute("SELECT inet_server_addr()")
result = cursor.fetchone()
print(f"Server address: {result[0]}")

cursor.close()
conn.close()

Install with Tessl CLI