tessl/pypi-psycopg2
Python-PostgreSQL Database Adapter
—
Pending
connections-cursors.mddocs/
Database Connections and Cursors
Core database connectivity functionality providing connection management, cursor operations, transaction handling, and the foundation for all database operations in psycopg2.
Capabilities
Connection Creation
Create database connections using connection strings or keyword parameters, with support for connection factories and default cursor types.
def connect(dsn=None, connection_factory=None, cursor_factory=None, **kwargs):
"""
Create a new database connection.
Parameters:
- dsn (str, optional): Connection string
- connection_factory (callable, optional): Custom connection class
- cursor_factory (callable, optional): Default cursor factory
- **kwargs: Connection parameters (host, port, database, user, password, etc.)
Returns:
connection: Database connection object
"""Usage Example:
import psycopg2
# Using connection string
conn = psycopg2.connect("host=localhost dbname=mydb user=myuser password=mypass")
# Using keyword arguments
conn = psycopg2.connect(
host="localhost",
port=5432,
database="mydb",
user="myuser",
password="mypass"
)
# With custom cursor factory
from psycopg2.extras import DictCursor
conn = psycopg2.connect(
host="localhost",
database="mydb",
user="myuser",
password="mypass",
cursor_factory=DictCursor
)Connection Management
Manage database connections with transaction control, configuration settings, and resource cleanup.
class connection:
"""Database connection object."""
def cursor(self, name=None, cursor_factory=None):
"""
Create a new cursor.
Parameters:
- name (str, optional): Server-side cursor name
- cursor_factory (callable, optional): Cursor class
Returns:
cursor: Database cursor object
"""
def commit(self):
"""Commit the current transaction."""
def rollback(self):
"""Rollback the current transaction."""
def close(self):
"""Close the connection."""
def set_isolation_level(self, level):
"""
Set transaction isolation level.
Parameters:
- level (int): Isolation level constant
"""
def set_client_encoding(self, encoding):
"""
Set client encoding.
Parameters:
- encoding (str): Encoding name
"""
def cancel(self):
"""Cancel current operation."""
def reset(self):
"""Reset connection to initial state."""
def set_session(self, isolation_level=None, readonly=None, deferrable=None, autocommit=None):
"""
Set session parameters.
Parameters:
- isolation_level (int, optional): Transaction isolation level
- readonly (bool, optional): Read-only mode
- deferrable (bool, optional): Deferrable transactions
- autocommit (bool, optional): Autocommit mode
"""
def get_transaction_status(self):
"""
Get backend transaction status.
Returns:
int: Transaction status constant
"""
# Properties
@property
def closed(self) -> int:
"""Connection status (0=open, >0=closed)."""
@property
def status(self) -> int:
"""Connection status constant."""
@property
def autocommit(self) -> bool:
"""Autocommit mode."""
@autocommit.setter
def autocommit(self, value: bool):
"""Set autocommit mode."""
@property
def isolation_level(self) -> int:
"""Current isolation level."""
@property
def encoding(self) -> str:
"""Client encoding."""
@property
def cursor_factory(self):
"""Default cursor factory."""
@cursor_factory.setter
def cursor_factory(self, factory):
"""Set default cursor factory."""Usage Example:
import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE
conn = psycopg2.connect("host=localhost dbname=mydb user=myuser")
# Transaction management
try:
with conn.cursor() as cur:
cur.execute("INSERT INTO users (name) VALUES (%s)", ("Alice",))
cur.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = %s", (1,))
conn.commit()
except Exception as e:
conn.rollback()
raise
# Configuration
conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
conn.autocommit = True
print(f"Connection status: {conn.status}")
print(f"Encoding: {conn.encoding}")
conn.close()Cursor Operations
Execute SQL queries and fetch results with support for parameterized queries, batch operations, and various result formats.
class cursor:
"""Database cursor object."""
def execute(self, query, vars=None):
"""
Execute a database operation.
Parameters:
- query (str): SQL query
- vars (sequence, optional): Query parameters
"""
def executemany(self, query, vars_list):
"""
Execute query multiple times with different parameters.
Parameters:
- query (str): SQL query
- vars_list (sequence): List of parameter tuples
"""
def fetchone(self):
"""
Fetch the next row of query result.
Returns:
tuple or None: Next row or None if no more rows
"""
def fetchmany(self, size=None):
"""
Fetch multiple rows of query result.
Parameters:
- size (int, optional): Number of rows to fetch
Returns:
list: List of row tuples
"""
def fetchall(self):
"""
Fetch all remaining rows of query result.
Returns:
list: List of all remaining row tuples
"""
def callproc(self, procname, parameters=None):
"""
Call a stored procedure.
Parameters:
- procname (str): Procedure name
- parameters (sequence, optional): Procedure parameters
Returns:
sequence: Modified parameters
"""
def close(self):
"""Close the cursor."""
def copy_from(self, file, table, sep='\t', null='\\N', size=8192, columns=None):
"""
Copy data from file to table.
Parameters:
- file: File-like object
- table (str): Table name
- sep (str): Field separator
- null (str): NULL representation
- size (int): Buffer size
- columns (sequence, optional): Column names
"""
def copy_to(self, file, table, sep='\t', null='\\N', columns=None):
"""
Copy data from table to file.
Parameters:
- file: File-like object
- table (str): Table name
- sep (str): Field separator
- null (str): NULL representation
- columns (sequence, optional): Column names
"""
def copy_expert(self, sql, file, size=8192):
"""
Execute COPY command with custom SQL.
Parameters:
- sql (str): COPY SQL command
- file: File-like object
- size (int): Buffer size
"""
def mogrify(self, operation, parameters=None):
"""
Return formatted query string.
Parameters:
- operation (str): SQL query
- parameters (sequence, optional): Query parameters
Returns:
bytes: Formatted query
"""
def scroll(self, value, mode='relative'):
"""
Move cursor position.
Parameters:
- value (int): Number of rows to move
- mode (str): 'relative' or 'absolute' positioning
"""
def setinputsizes(self, sizes):
"""Set input parameter sizes (no-op in psycopg2)."""
def setoutputsize(self, size, column=None):
"""Set output column size (no-op in psycopg2)."""
# Properties
@property
def description(self):
"""Column information for last query."""
@property
def rowcount(self) -> int:
"""Number of rows affected by last operation."""
@property
def rownumber(self) -> int:
"""Current row number."""
@property
def lastrowid(self):
"""Last inserted row ID (PostgreSQL doesn't support this)."""
@property
def query(self):
"""Last executed query."""
@property
def statusmessage(self) -> str:
"""Status message from last operation."""
@property
def connection(self):
"""Connection this cursor belongs to."""
@property
def name(self) -> str:
"""Server-side cursor name."""
@property
def scrollable(self) -> bool:
"""Whether cursor is scrollable."""
@property
def withhold(self) -> bool:
"""Whether cursor is WITH HOLD."""Usage Example:
import psycopg2
conn = psycopg2.connect("host=localhost dbname=mydb user=myuser")
# Basic query execution
with conn.cursor() as cur:
cur.execute("SELECT id, name, email FROM users WHERE age > %s", (25,))
# Fetch results
row = cur.fetchone()
while row:
print(f"ID: {row[0]}, Name: {row[1]}, Email: {row[2]}")
row = cur.fetchone()
# Batch operations
with conn.cursor() as cur:
users = [
("Alice", "alice@example.com", 28),
("Bob", "bob@example.com", 32),
("Charlie", "charlie@example.com", 24)
]
cur.executemany(
"INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
users
)
print(f"Inserted {cur.rowcount} rows")
# COPY operations
import io
data = "1\tAlice\talice@example.com\n2\tBob\tbob@example.com\n"
with conn.cursor() as cur:
cur.copy_from(
io.StringIO(data),
'users',
columns=('id', 'name', 'email'),
sep='\t'
)
conn.commit()
conn.close()Server-Side Cursors
Named cursors that execute on the PostgreSQL server, enabling efficient processing of large result sets without loading all data into memory.
# Server-side cursor creation
def cursor(name, cursor_factory=None, scrollable=None, withhold=False):
"""
Create named server-side cursor.
Parameters:
- name (str): Cursor name
- cursor_factory (callable, optional): Cursor class
- scrollable (bool, optional): Enable scrolling
- withhold (bool): Preserve cursor after transaction
Returns:
cursor: Server-side cursor
"""Usage Example:
import psycopg2
conn = psycopg2.connect("host=localhost dbname=mydb user=myuser")
# Server-side cursor for large result sets
with conn.cursor("large_query") as cur:
cur.execute("SELECT * FROM large_table ORDER BY id")
# Fetch in chunks
while True:
rows = cur.fetchmany(1000)
if not rows:
break
for row in rows:
# Process row without loading entire result set
process_row(row)
conn.close()Context Managers
Automatic resource management using Python context managers for connections and cursors.
Usage Example:
import psycopg2
# Connection context manager
with psycopg2.connect("host=localhost dbname=mydb user=myuser") as conn:
with conn.cursor() as cur:
cur.execute("SELECT * FROM users")
users = cur.fetchall()
# Cursor automatically closed
# Transaction automatically committed if no exception,
# or rolled back if exception occurs
# Connection automatically closedTypes
Connection Status Constants
STATUS_SETUP: int # 0 - Connection being set up
STATUS_READY: int # 1 - Connection ready for queries
STATUS_BEGIN: int # 2 - Transaction in progress
STATUS_SYNC: int # 3 - Synchronizing connection
STATUS_ASYNC: int # 4 - Asynchronous connection
STATUS_PREPARED: int # 5 - Prepared for async operation
STATUS_IN_TRANSACTION: int # Alias for STATUS_BEGINTransaction Status Constants
TRANSACTION_STATUS_IDLE: int # 0 - Outside transaction
TRANSACTION_STATUS_ACTIVE: int # 1 - Command in progress
TRANSACTION_STATUS_INTRANS: int # 2 - In transaction block
TRANSACTION_STATUS_INERROR: int # 3 - In failed transaction
TRANSACTION_STATUS_UNKNOWN: int # 4 - Connection badIsolation Level Constants
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
ISOLATION_LEVEL_DEFAULT: None # NoneInstall with Tessl CLI
npx tessl i tessl/pypi-psycopg2