tessl/pypi-py-postgresql
PostgreSQL driver and tools library providing PG-API and DB-API 2.0 interfaces for Python.
—
Pending
query-execution.mddocs/
Query Execution
Prepared statement interface with parameter binding, result streaming, and transaction management for efficient and secure database operations.
Capabilities
Statement Preparation
Create prepared statements for efficient repeated execution with automatic parameter binding and type conversion.
def prepare(statement):
"""
Create a prepared statement from SQL with parameter placeholders.
Parameters:
- statement (str): SQL statement with $1, $2, etc. parameter placeholders
Returns:
Statement: Prepared statement object for repeated execution
Raises:
QueryError: If statement cannot be prepared
"""Statement Interface
Prepared statement object providing multiple execution modes for different use cases.
class Statement:
"""
Prepared statement with parameter binding and multiple execution modes.
"""
def __call__(*parameters):
"""
Execute statement and return all results.
Parameters:
- *parameters: Values for statement parameters ($1, $2, etc.)
Returns:
List of result rows or command result
Raises:
QueryError: If execution fails
"""
def first(*parameters):
"""
Execute statement and return first result row.
Parameters:
- *parameters: Values for statement parameters
Returns:
Row or None: First result row or None if no results
"""
def rows(*parameters):
"""
Execute statement and return iterator over result rows.
Parameters:
- *parameters: Values for statement parameters
Returns:
Iterator[Row]: Iterator over result rows for streaming
"""
def chunks(*parameters):
"""
Execute statement and return chunked results.
Parameters:
- *parameters: Values for statement parameters
Returns:
Chunks: Chunked result iterator for large datasets
"""
def column(*parameters):
"""
Execute statement and return iterator over single column values.
Parameters:
- *parameters: Values for statement parameters
Returns:
Iterator: Iterator over values from the first column
"""
def declare(*parameters):
"""
Create a scrollable cursor for the statement.
Parameters:
- *parameters: Values for statement parameters
Returns:
Cursor: Scrollable cursor for navigation and fetching
"""
def close():
"""Close the prepared statement."""
@property
def sql():
"""
Get the SQL text of the prepared statement.
Returns:
str: Original SQL statement text
"""
@property
def parameter_types():
"""
Get parameter type information.
Returns:
List[int]: PostgreSQL type OIDs for parameters
"""
@property
def result_types():
"""
Get result column type information.
Returns:
List[int]: PostgreSQL type OIDs for result columns
"""Direct Query Execution
Execute queries directly without preparation for one-time operations.
def execute(statement, *parameters):
"""
Execute a statement directly with parameters.
Parameters:
- statement (str): SQL statement
- *parameters: Parameter values
Returns:
Command result or row data
"""
def query(statement, *parameters):
"""
Execute a query and return all results.
Parameters:
- statement (str): SQL query
- *parameters: Parameter values
Returns:
List[Row]: All result rows
"""Result Row Interface
Result rows providing both positional and named access to column values.
class Row:
"""
Result row with named and positional access to column values.
"""
def __getitem__(key):
"""
Get column value by index or name.
Parameters:
- key (int or str): Column index or name
Returns:
Column value with automatic type conversion
"""
def __len__():
"""
Get number of columns in row.
Returns:
int: Number of columns
"""
def keys():
"""
Get column names.
Returns:
List[str]: Column names
"""
def values():
"""
Get column values.
Returns:
List: Column values
"""
def items():
"""
Get column name-value pairs.
Returns:
List[tuple]: (name, value) pairs
"""Chunked Results
Interface for processing large result sets in chunks to manage memory usage.
class Chunks:
"""
Chunked result iterator for processing large datasets efficiently.
"""
def __iter__():
"""
Iterate over result chunks.
Returns:
Iterator[List[Row]]: Iterator over chunks of rows
"""
def __next__():
"""
Get next chunk of results.
Returns:
List[Row]: Next chunk of rows
Raises:
StopIteration: When no more chunks available
"""
def close():
"""Close the chunked result iterator."""Cursor Interface
Scrollable cursor for bidirectional navigation through result sets with seek operations.
class Cursor:
"""
Scrollable cursor providing bidirectional navigation through query results.
"""
def read(quantity=None, direction=None):
"""
Read rows from the cursor position.
Parameters:
- quantity (int, optional): Number of rows to read (default: all remaining)
- direction (str, optional): 'FORWARD' or 'BACKWARD' (default: cursor direction)
Returns:
List[Row]: List of rows read from cursor
"""
def seek(offset, whence='ABSOLUTE'):
"""
Move cursor to specified position.
Parameters:
- offset (int): Position offset
- whence (str): 'ABSOLUTE', 'RELATIVE', 'FORWARD', or 'BACKWARD'
"""
def __next__():
"""
Get next row from cursor.
Returns:
Row: Next row in cursor direction
Raises:
StopIteration: When no more rows available
"""
def clone():
"""
Create a copy of the cursor.
Returns:
Cursor: New cursor instance at same position
"""
def close():
"""Close the cursor and release resources."""
@property
def direction():
"""
Get cursor direction.
Returns:
bool: True for FORWARD, False for BACKWARD
"""
@property
def cursor_id():
"""
Get cursor identifier.
Returns:
str: Unique cursor identifier
"""
@property
def column_names():
"""
Get result column names.
Returns:
List[str]: Column names in result order
"""
@property
def column_types():
"""
Get result column types.
Returns:
List[type]: Python types for result columns
"""
@property
def statement():
"""
Get associated statement.
Returns:
Statement: The statement that created this cursor
"""Usage Examples
Basic Statement Preparation and Execution
import postgresql
db = postgresql.open('pq://user:pass@localhost/mydb')
# Prepare statement for repeated use
get_user = db.prepare("SELECT id, name, email FROM users WHERE id = $1")
# Execute with parameter
user = get_user.first(123)
if user:
print(f"User: {user['name']} ({user['email']})")
# Execute multiple times efficiently
user_ids = [1, 2, 3, 4, 5]
for user_id in user_ids:
user = get_user.first(user_id)
if user:
print(f"ID {user_id}: {user['name']}")
get_user.close()Streaming Large Result Sets
import postgresql
db = postgresql.open('pq://user:pass@localhost/mydb')
# Prepare query for large dataset
get_all_orders = db.prepare("""
SELECT order_id, customer_id, order_date, total
FROM orders
WHERE order_date >= $1
ORDER BY order_date
""")
# Stream results to avoid loading all into memory
from datetime import date
start_date = date(2023, 1, 1)
total_amount = 0
order_count = 0
for order in get_all_orders.rows(start_date):
total_amount += order['total']
order_count += 1
# Process order
print(f"Order {order['order_id']}: ${order['total']}")
print(f"Processed {order_count} orders, total: ${total_amount}")Chunked Processing
import postgresql
db = postgresql.open('pq://user:pass@localhost/mydb')
# Process large dataset in chunks
get_transactions = db.prepare("SELECT * FROM transactions WHERE processed = false")
# Process in chunks of 1000 rows
for chunk in get_transactions.chunks():
batch_updates = []
for transaction in chunk:
# Process transaction
result = process_transaction(transaction)
batch_updates.append((result, transaction['id']))
# Batch update processed transactions
update_stmt = db.prepare("UPDATE transactions SET result = $1, processed = true WHERE id = $2")
for result, tx_id in batch_updates:
update_stmt(result, tx_id)
print(f"Processed chunk of {len(chunk)} transactions")Parameter Types and Query Building
import postgresql
db = postgresql.open('pq://user:pass@localhost/mydb')
# Prepare statement with multiple parameter types
search_query = db.prepare("""
SELECT product_id, name, price, in_stock
FROM products
WHERE category = $1
AND price BETWEEN $2 AND $3
AND in_stock = $4
ORDER BY price
""")
# Execute with different parameter types
products = search_query(
"electronics", # str
100.0, # float
500.0, # float
True # bool
)
for product in products:
stock_status = "In Stock" if product['in_stock'] else "Out of Stock"
print(f"{product['name']}: ${product['price']} - {stock_status}")
# Check statement metadata
print(f"Parameter types: {search_query.parameter_types}")
print(f"Result types: {search_query.result_types}")Direct Query Execution
import postgresql
db = postgresql.open('pq://user:pass@localhost/mydb')
# One-time queries don't need preparation
table_count = db.query("SELECT count(*) FROM information_schema.tables WHERE table_schema = 'public'")[0][0]
print(f"Public tables: {table_count}")
# Execute DDL or commands
db.execute("CREATE INDEX IF NOT EXISTS idx_user_email ON users(email)")
db.execute("ANALYZE users")
# Execute with parameters
recent_users = db.query(
"SELECT name FROM users WHERE created_at > $1",
datetime.now() - timedelta(days=7)
)
for user in recent_users:
print(f"Recent user: {user['name']}")Error Handling
import postgresql
import postgresql.exceptions as pg_exc
db = postgresql.open('pq://user:pass@localhost/mydb')
try:
# Prepare potentially problematic statement
stmt = db.prepare("SELECT * FROM maybe_missing_table WHERE id = $1")
result = stmt.first(123)
except pg_exc.ProgrammingError as e:
print(f"SQL error: {e}")
# Handle missing table, invalid SQL, etc.
except pg_exc.DataError as e:
print(f"Data error: {e}")
# Handle invalid parameter values, type conversion errors
except pg_exc.ConnectionError as e:
print(f"Connection error: {e}")
# Handle connection issues
finally:
if 'stmt' in locals():
stmt.close()Install with Tessl CLI
npx tessl i tessl/pypi-py-postgresql