tessl/pypi-peewee
A lightweight Python ORM (Object-Relational Mapping) library that provides a simple and expressive interface for database operations.
—
Pending
database-and-connections.mddocs/
Database and Connections
Database connection management, transaction handling, and backend-specific functionality. Peewee supports multiple database backends with consistent APIs while providing backend-specific optimizations and features.
Capabilities
Database Base Class
Core database functionality providing connection management, transaction control, and SQL execution capabilities shared across all database backends.
class Database:
"""
Base database class providing core functionality.
"""
def __init__(self, database, **kwargs):
"""
Initialize database connection.
Parameters:
- database (str): Database name/path
- **kwargs: Backend-specific connection parameters
"""
def connect(self, reuse_if_open=False):
"""
Connect to the database.
Parameters:
- reuse_if_open (bool): Reuse existing connection
Returns:
Connection: Database connection object
"""
def close(self):
"""
Close the database connection.
Returns:
bool: True if connection was closed
"""
def is_closed(self):
"""
Check if database connection is closed.
Returns:
bool: True if connection is closed
"""
def connection(self):
"""
Get the current database connection.
Returns:
Connection: Current connection object
"""
def execute_sql(self, sql, params=None, commit=None):
"""
Execute raw SQL query.
Parameters:
- sql (str): SQL query string
- params (list): Query parameters
- commit (bool): Auto-commit transaction
Returns:
Cursor: Database cursor with results
"""
def begin(self):
"""
Begin a database transaction.
"""
def commit(self):
"""
Commit the current transaction.
"""
def rollback(self):
"""
Rollback the current transaction.
"""
def atomic(self, *args, **kwargs):
"""
Create atomic transaction context manager.
Returns:
Context manager for atomic operations
"""
def transaction(self, *args, **kwargs):
"""
Create transaction context manager.
Returns:
Context manager for transactions
"""
def savepoint(self, sid=None):
"""
Create savepoint context manager.
Parameters:
- sid (str): Savepoint identifier
Returns:
Context manager for savepoint
"""
def create_tables(self, models, **options):
"""
Create database tables for models.
Parameters:
- models (list): Model classes to create tables for
- **options: Creation options (safe, checkfirst, etc.)
"""
def drop_tables(self, models, **options):
"""
Drop database tables for models.
Parameters:
- models (list): Model classes to drop tables for
- **options: Drop options (safe, cascade, etc.)
"""
def truncate_tables(self, models, **options):
"""
Truncate database tables for models.
Parameters:
- models (list): Model classes to truncate
- **options: Truncate options
"""
def create_index(self, model_class, fields, unique=False, **kwargs):
"""
Create database index.
Parameters:
- model_class: Model class
- fields (list): Fields to index
- unique (bool): Create unique index
- **kwargs: Index options
"""
def drop_index(self, model_class, fields, **kwargs):
"""
Drop database index.
Parameters:
- model_class: Model class
- fields (list): Fields indexed
- **kwargs: Drop options
"""SQLite Database
SQLite database implementation providing file-based storage with extensive SQL support and performance optimizations.
class SqliteDatabase(Database):
"""
SQLite database implementation.
"""
def __init__(self, database, **kwargs):
"""
Initialize SQLite database.
Parameters:
- database (str): Database file path (':memory:' for in-memory)
- pragmas (dict): SQLite PRAGMA settings
- timeout (int): Connection timeout in seconds
- check_same_thread (bool): Check thread safety
- isolation_level (str): Transaction isolation level
- **kwargs: Additional sqlite3 connection parameters
"""
def pragma(self, key, value=None, permanent=False):
"""
Get or set SQLite PRAGMA value.
Parameters:
- key (str): PRAGMA name
- value: PRAGMA value (None to get current)
- permanent (bool): Apply to all connections
Returns:
Current value if getting, None if setting
"""
def attach(self, database, name):
"""
Attach another database.
Parameters:
- database (str): Database file path
- name (str): Alias name for attached database
"""
def detach(self, name):
"""
Detach database.
Parameters:
- name (str): Alias name of attached database
"""Usage example:
from peewee import *
# Basic SQLite database
db = SqliteDatabase('app.db')
# In-memory database
memory_db = SqliteDatabase(':memory:')
# SQLite with pragmas
db = SqliteDatabase('app.db', pragmas={
'journal_mode': 'wal',
'cache_size': -64000, # 64MB cache
'foreign_keys': 1,
'ignore_check_constraints': 0,
'synchronous': 0
})
# Connect and use
db.connect()
db.pragma('foreign_keys', 1) # Enable foreign keys
db.close()PostgreSQL Database
PostgreSQL database implementation with support for advanced SQL features, JSON, arrays, and PostgreSQL-specific functionality.
class PostgresqlDatabase(Database):
"""
PostgreSQL database implementation.
"""
def __init__(self, database, **kwargs):
"""
Initialize PostgreSQL database.
Parameters:
- database (str): Database name
- user (str): Username
- password (str): Password
- host (str): Host address (default: localhost)
- port (int): Port number (default: 5432)
- sslmode (str): SSL mode ('disable', 'require', etc.)
- **kwargs: Additional psycopg2 connection parameters
"""
def set_time_zone(self, timezone):
"""
Set session timezone.
Parameters:
- timezone (str): Timezone name
"""Usage example:
from peewee import *
# PostgreSQL database
db = PostgresqlDatabase(
'mydb',
user='postgres',
password='secret',
host='localhost',
port=5432
)
# PostgreSQL with SSL
db = PostgresqlDatabase(
'mydb',
user='postgres',
password='secret',
host='db.example.com',
sslmode='require'
)MySQL Database
MySQL/MariaDB database implementation with support for MySQL-specific features and optimizations.
class MySQLDatabase(Database):
"""
MySQL/MariaDB database implementation.
"""
def __init__(self, database, **kwargs):
"""
Initialize MySQL database.
Parameters:
- database (str): Database name
- user (str): Username
- password (str): Password
- host (str): Host address (default: localhost)
- port (int): Port number (default: 3306)
- charset (str): Character set (default: utf8mb4)
- sql_mode (str): SQL mode settings
- **kwargs: Additional connection parameters
"""Usage example:
from peewee import *
# MySQL database
db = MySQLDatabase(
'mydb',
user='root',
password='secret',
host='localhost',
port=3306,
charset='utf8mb4'
)Database Proxy
Database proxy for late binding, allowing models to be defined before the actual database instance is available.
class DatabaseProxy:
"""
Proxy object for late database binding.
"""
def initialize(self, database):
"""
Initialize with actual database instance.
Parameters:
- database: Database instance to proxy to
"""
def __getattr__(self, attr):
"""
Proxy attribute access to underlying database.
"""
def __setattr__(self, attr, value):
"""
Proxy attribute setting to underlying database.
"""Usage example:
from peewee import *
# Define proxy
database = DatabaseProxy()
class User(Model):
username = CharField()
class Meta:
database = database
# Later, initialize with actual database
actual_db = SqliteDatabase('app.db')
database.initialize(actual_db)Base Proxy Class
Generic proxy implementation that DatabaseProxy extends.
class Proxy:
"""
Base proxy class for creating placeholders for objects.
"""
def __init__(self):
"""Initialize empty proxy with callback support."""
def initialize(self, obj):
"""
Initialize proxy with target object.
Parameters:
- obj: Object to proxy to
"""
def attach_callback(self, callback):
"""
Attach callback to be called when proxy is initialized.
Parameters:
- callback: Function to call with target object
"""Transaction Management
Context managers and decorators for managing database transactions, atomicity, and savepoints.
def atomic(db=None):
"""
Decorator/context manager for atomic operations.
Parameters:
- db: Database instance (uses default if None)
Usage as decorator:
@atomic()
def transfer_funds(from_account, to_account, amount):
...
Usage as context manager:
with atomic():
...
"""
def transaction(db=None):
"""
Decorator/context manager for transactions.
Parameters:
- db: Database instance (uses default if None)
"""
def savepoint(db=None, sid=None):
"""
Create savepoint context manager.
Parameters:
- db: Database instance
- sid (str): Savepoint identifier
"""Usage examples:
from peewee import *
db = SqliteDatabase('app.db')
# Atomic decorator
@atomic(db)
def create_user_with_profile(username, email, bio):
user = User.create(username=username, email=email)
UserProfile.create(user=user, bio=bio)
return user
# Atomic context manager
with atomic(db):
user = User.create(username='john', email='john@example.com')
user.is_active = True
user.save()
# Nested transactions with savepoints
with transaction(db):
user = User.create(username='jane', email='jane@example.com')
with savepoint(db):
# This might fail
try:
UserProfile.create(user=user, bio='A very long bio...')
except IntegrityError:
# Rollback to savepoint, user creation still valid
pass
user.is_active = True
user.save()Connection Management
Advanced connection handling, pooling configuration, and connection lifecycle management.
class Database:
def init(self, database, **kwargs):
"""
Initialize database with new parameters.
Parameters:
- database (str): Database identifier
- **kwargs: Connection parameters
"""
def bind(self, models, bind_refs=True, bind_backrefs=True):
"""
Bind models to this database.
Parameters:
- models (list): Model classes to bind
- bind_refs (bool): Bind foreign key references
- bind_backrefs (bool): Bind back-references
"""
def bind_ctx(self, models, bind_refs=True, bind_backrefs=True):
"""
Context manager for temporary model binding.
Parameters:
- models (list): Model classes to bind
- bind_refs (bool): Bind foreign key references
- bind_backrefs (bool): Bind back-references
"""Schema Management
Database schema operations including table creation, modification, and introspection.
class SchemaManager:
"""
Database schema management and DDL operations.
"""
def __init__(self, database):
"""
Parameters:
- database: Database instance to manage
"""
def create_table(self, model_class, safe=True, **options):
"""
Create table for model.
Parameters:
- model_class: Model to create table for
- safe (bool): Don't raise error if table exists
- **options: Creation options
"""
def drop_table(self, model_class, safe=True, cascade=False, **options):
"""
Drop table for model.
Parameters:
- model_class: Model to drop table for
- safe (bool): Don't raise error if table doesn't exist
- cascade (bool): Drop dependent objects
- **options: Drop options
"""
def truncate_table(self, model_class, restart_identity=False, cascade=False):
"""
Truncate table for model.
Parameters:
- model_class: Model to truncate
- restart_identity (bool): Restart sequence counters
- cascade (bool): Truncate dependent tables
"""
def create_index(self, model_class, fields, unique=False, safe=True, **kwargs):
"""
Create index on model fields.
Parameters:
- model_class: Model class
- fields (list): Fields to index
- unique (bool): Create unique index
- safe (bool): Don't raise error if index exists
- **kwargs: Index options
"""
def drop_index(self, model_class, fields, **kwargs):
"""
Drop index from model fields.
Parameters:
- model_class: Model class
- fields (list): Fields indexed
- **kwargs: Drop options
"""
def create_foreign_key(self, model_class, field, constraint=None):
"""
Create foreign key constraint.
Parameters:
- model_class: Model class
- field: Foreign key field
- constraint (str): Constraint name
"""
def create_sequence(self, sequence_name):
"""
Create database sequence.
Parameters:
- sequence_name (str): Name of sequence to create
"""
def drop_sequence(self, sequence_name):
"""
Drop database sequence.
Parameters:
- sequence_name (str): Name of sequence to drop
"""
class Index:
"""
Database index definition.
"""
def __init__(self, name, table, expressions, unique=False, safe=True, where=None, **kwargs):
"""
Parameters:
- name (str): Index name
- table (str): Table name
- expressions (list): Column expressions
- unique (bool): Unique constraint
- safe (bool): Create if not exists
- where: Partial index condition
- **kwargs: Database-specific options
"""
class ModelIndex:
"""
Model-specific index definition.
"""
def __init__(self, model, fields, unique=False, **kwargs):
"""
Parameters:
- model: Model class
- fields (list): Field names or expressions
- unique (bool): Unique constraint
- **kwargs: Index options
"""Error Handling
Database-specific error handling and exception management:
from peewee import *
try:
db.connect()
# Database operations
except OperationalError as e:
print(f"Database connection failed: {e}")
except ProgrammingError as e:
print(f"SQL syntax error: {e}")
except IntegrityError as e:
print(f"Constraint violation: {e}")
finally:
if not db.is_closed():
db.close()Install with Tessl CLI
npx tessl i tessl/pypi-peewee