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
legacy-dbapi.mddocs/
Legacy DB-API 2.0 Interface
Standard Python Database API 2.0 compliant interface providing Connection and Cursor classes for traditional database programming patterns with full transaction support.
Capabilities
Connection Class
DB-API 2.0 compliant connection class providing transaction management, cursor creation, and prepared statement support.
class Connection:
"""
DB-API 2.0 compliant database connection.
Properties:
- autocommit: bool - Enable/disable automatic transaction commits
- description: tuple - Column descriptions from last query
- _in_transaction: bool - Whether currently in transaction
"""
def cursor(self) -> Cursor:
"""
Create a new cursor for executing queries.
Returns:
New Cursor instance bound to this connection
"""
def commit(self) -> None:
"""
Commit the current transaction.
Raises:
DatabaseError: If commit fails
"""
def rollback(self) -> None:
"""
Roll back the current transaction.
Raises:
DatabaseError: If rollback fails
"""
def run(self, sql: str, stream=None, **params) -> tuple:
"""
Execute SQL statement with named parameters.
Parameters:
- sql: SQL statement with :param placeholders
- stream: Optional stream for COPY operations
- **params: Named parameters for SQL statement
Returns:
Tuple of result rows
Raises:
ProgrammingError: Invalid SQL syntax
DataError: Invalid data values
"""
def prepare(self, operation: str) -> PreparedStatement:
"""
Create a prepared statement for efficient repeated execution.
Parameters:
- operation: SQL statement to prepare
Returns:
PreparedStatement object
Raises:
ProgrammingError: Invalid SQL syntax
"""
def close(self) -> None:
"""
Close the database connection.
"""Cursor Class
DB-API 2.0 compliant cursor for executing queries and fetching results with support for parameterized queries and result iteration.
class Cursor:
"""
DB-API 2.0 compliant cursor for query execution.
Properties:
- arraysize: int - Number of rows to fetch with fetchmany()
- description: tuple - Column descriptions from last query
- rowcount: int - Number of rows affected by last query
- connection: Connection - Parent connection object
"""
def execute(self, operation: str, args: tuple = (), stream=None) -> Cursor:
"""
Execute SQL statement with positional parameters.
Parameters:
- operation: SQL statement with %s placeholders
- args: Tuple of parameter values
- stream: Optional stream for COPY operations
Returns:
Self for method chaining
Raises:
ProgrammingError: Invalid SQL syntax
DataError: Invalid parameter values
"""
def executemany(self, operation: str, param_sets: list) -> Cursor:
"""
Execute SQL statement multiple times with different parameters.
Parameters:
- operation: SQL statement with %s placeholders
- param_sets: List of parameter tuples
Returns:
Self for method chaining
Raises:
ProgrammingError: Invalid SQL syntax
DataError: Invalid parameter values
"""
def fetchone(self) -> tuple:
"""
Fetch the next row from query results.
Returns:
Single row tuple or None if no more rows
Raises:
Error: If no query has been executed
"""
def fetchmany(self, num: int = None) -> tuple:
"""
Fetch multiple rows from query results.
Parameters:
- num: Number of rows to fetch (defaults to arraysize)
Returns:
Tuple of row tuples
Raises:
Error: If no query has been executed
"""
def fetchall(self) -> tuple:
"""
Fetch all remaining rows from query results.
Returns:
Tuple of all remaining row tuples
Raises:
Error: If no query has been executed
"""
def close(self) -> None:
"""
Close the cursor and free associated resources.
"""
def setinputsizes(self, *sizes) -> None:
"""
Set input parameter sizes (no-op for compatibility).
Parameters:
- *sizes: Parameter size specifications (ignored)
"""
def setoutputsize(self, size: int, column: int = None) -> None:
"""
Set output column size (no-op for compatibility).
Parameters:
- size: Column size specification (ignored)
- column: Column index (ignored)
"""PreparedStatement Class
Prepared statement class for efficient repeated execution of SQL statements with different parameter values.
class PreparedStatement:
"""
Prepared statement for efficient repeated execution.
"""
def run(self, **vals) -> tuple:
"""
Execute prepared statement with named parameters.
Parameters:
- **vals: Named parameter values
Returns:
Tuple of result rows
Raises:
DataError: Invalid parameter values
OperationalError: Execution errors
"""
def close(self) -> None:
"""
Close the prepared statement and free resources.
"""Two-Phase Commit Support
Support for distributed transactions using the Two-Phase Commit protocol.
class Connection:
def xid(self, format_id: int, global_transaction_id: str, branch_qualifier: str) -> tuple:
"""
Create transaction identifier for two-phase commit.
Parameters:
- format_id: Format identifier
- global_transaction_id: Global transaction identifier
- branch_qualifier: Branch qualifier
Returns:
Transaction identifier tuple
"""
def tpc_begin(self, xid: tuple) -> None:
"""
Begin two-phase commit transaction.
Parameters:
- xid: Transaction identifier from xid()
Raises:
NotSupportedError: If two-phase commit not supported
"""
def tpc_prepare(self) -> None:
"""
Prepare two-phase commit transaction.
Raises:
DatabaseError: If prepare fails
"""
def tpc_commit(self, xid: tuple = None) -> None:
"""
Commit two-phase commit transaction.
Parameters:
- xid: Transaction identifier (optional)
Raises:
DatabaseError: If commit fails
"""
def tpc_rollback(self, xid: tuple = None) -> None:
"""
Roll back two-phase commit transaction.
Parameters:
- xid: Transaction identifier (optional)
Raises:
DatabaseError: If rollback fails
"""
def tpc_recover(self) -> list:
"""
Get list of pending transaction identifiers.
Returns:
List of pending transaction identifier tuples
"""DB-API 2.0 Constructor Functions
Standard constructor functions required by the DB-API 2.0 specification for creating date/time objects.
def Date(year: int, month: int, day: int) -> datetime.date:
"""
Construct a date object (alias for PgDate).
This function is part of the DBAPI 2.0 specification.
"""
def Time(hour: int, minute: int, second: int) -> datetime.time:
"""
Construct a time object (alias for PgTime).
This function is part of the DBAPI 2.0 specification.
"""
def Timestamp(year: int, month: int, day: int, hour: int, minute: int, second: int) -> datetime.datetime:
"""
Construct a timestamp object.
This function is part of the DBAPI 2.0 specification.
"""
def DateFromTicks(ticks: float) -> datetime.date:
"""
Construct a date object from seconds since epoch.
This function is part of the DBAPI 2.0 specification.
"""
def TimeFromTicks(ticks: float) -> datetime.time:
"""
Construct a time object from seconds since epoch.
This function is part of the DBAPI 2.0 specification.
"""
def TimestampFromTicks(ticks: float) -> datetime.datetime:
"""
Construct a timestamp object from seconds since epoch.
This function is part of the DBAPI 2.0 specification.
"""
def Binary(value: bytes) -> bytes:
"""
Construct a binary data object.
This function is part of the DBAPI 2.0 specification.
"""Usage Examples
Basic Query Execution
import pg8000
# Connect and create cursor
conn = pg8000.connect(user="myuser", password="mypass", database="mydb")
cursor = conn.cursor()
# Execute simple query
cursor.execute("SELECT id, name FROM users WHERE active = %s", (True,))
# Fetch results
while True:
row = cursor.fetchone()
if row is None:
break
print(f"ID: {row[0]}, Name: {row[1]}")
# Clean up
cursor.close()
conn.close()Transaction Management
import pg8000
conn = pg8000.connect(user="myuser", password="mypass", database="mydb")
cursor = conn.cursor()
try:
# Start transaction (implicit)
cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)",
("John Doe", "john@example.com"))
cursor.execute("UPDATE user_stats SET count = count + 1")
# Commit transaction
conn.commit()
print("Transaction committed successfully")
except pg8000.DatabaseError as e:
# Roll back on error
conn.rollback()
print(f"Transaction rolled back: {e}")
finally:
cursor.close()
conn.close()Prepared Statements
import pg8000
conn = pg8000.connect(user="myuser", password="mypass", database="mydb")
# Prepare statement
stmt = conn.prepare("INSERT INTO logs (level, message, timestamp) VALUES (:level, :msg, :ts)")
# Execute multiple times with different parameters
import datetime
for level, message in [("INFO", "App started"), ("ERROR", "Database error"), ("INFO", "Request processed")]:
result = stmt.run(
level=level,
msg=message,
ts=datetime.datetime.now()
)
# Clean up
stmt.close()
conn.close()Install with Tessl CLI
npx tessl i tessl/pypi-pg8000