tessl/pypi-pony
Pony Object-Relational Mapper for Python with Pythonic query syntax using generator expressions
—
Pending
database-entities.mddocs/
Database and Entity Management
Core classes for defining database schemas and entity relationships. The Database class serves as the central registry and connection manager, while Entity provides the base for all ORM-mapped classes with automatic query generation capabilities.
Capabilities
Database Class
The Database class is the central component that manages entity definitions, database connections, and schema generation. Each Database instance represents a complete database schema.
class Database:
def __init__(self, *args, **kwargs):
"""Initialize database connection configuration.
Args:
*args: Positional arguments - first arg can be provider name or dict of connection params
**kwargs: Connection parameters including:
provider: Database provider ('sqlite', 'mysql', 'postgresql', 'oracle', 'cockroach')
filename: Database file path (SQLite only)
host: Database host (network databases)
port: Database port (network databases)
user: Database username (network databases)
password: Database password (network databases)
database: Database name (network databases)
"""
def bind(self, provider, **kwargs):
"""Bind database to specific provider and connection parameters.
Args:
provider: Database provider name
**kwargs: Connection parameters (host, port, user, password, etc.)
"""
def generate_mapping(self, check_tables=True, create_tables=False):
"""Generate database mapping from entity definitions.
Args:
check_tables: Validate existing table structures
create_tables: Create missing tables automatically
"""
def create_tables(self):
"""Create all tables for defined entities."""
def drop_all_tables(self, with_all_data=False):
"""Drop all tables from database.
Args:
with_all_data: If True, drop tables even if they contain data
"""
def drop_table(self, table_name, if_exists=False, with_all_data=False):
"""Drop specific table from database.
Args:
table_name: Name of table to drop
if_exists: Don't raise error if table doesn't exist
with_all_data: Drop table even if it contains data
"""
def check_tables(self):
"""Check that all entity tables exist and have correct structure."""
def get(self, sql, globals=None, locals=None):
"""Execute SQL query and get results.
Args:
sql: SQL query string or AST
globals: Global variables for SQL execution
locals: Local variables for SQL execution
Returns:
Query results
"""
def exists(self, sql, globals=None, locals=None):
"""Check if SQL query returns any results.
Args:
sql: SQL query string or AST
globals: Global variables for SQL execution
locals: Local variables for SQL execution
Returns:
bool: True if query returns results
"""
def insert(self, table_name, returning=None, **kwargs):
"""Insert row into table.
Args:
table_name: Name of table
returning: Column to return after insert
**kwargs: Column values to insert
Returns:
Inserted row ID or specified returning value
"""
def set_perms_for(self, *entities):
"""Set permissions for specified entities.
Args:
*entities: Entity classes to set permissions for
"""
# Properties
@property
def last_sql(self):
"""Get the last SQL query executed (str)."""
@property
def local_stats(self):
"""Get local query statistics (dict)."""
@property
def global_stats(self):
"""Get global query statistics (dict)."""Entity Base Class
The Entity metaclass provides the foundation for all database-mapped classes. Entities automatically receive query methods and database interaction capabilities.
class Entity:
_database_: Database # Associated database instance
_table_: str # Database table name
_pk_columns_: List[str] # Primary key column names
_pk_val_: Any # Primary key value
def __init__(self, **kwargs):
"""Create new entity instance.
Args:
**kwargs: Attribute values for entity creation
"""
def delete(self):
"""Delete this entity from database."""
def flush(self):
"""Flush changes to database without committing transaction."""
def get_pk(self):
"""Get primary key value.
Returns:
Primary key value (single value or tuple for composite keys)
"""
def set(self, **kwargs):
"""Update entity attributes.
Args:
**kwargs: Attribute values to update
"""
def __lt__(self, other):
"""Less than comparison based on primary key."""
def __le__(self, other):
"""Less than or equal comparison based on primary key."""
def __gt__(self, other):
"""Greater than comparison based on primary key."""
def __ge__(self, other):
"""Greater than or equal comparison based on primary key."""
@classmethod
def get(cls, *args, **kwargs):
"""Get single entity by primary key or conditions.
Args:
*args: Primary key values
**kwargs: Attribute conditions
Returns:
Entity instance
Raises:
ObjectNotFound: No entity matches criteria
MultipleObjectsFoundError: Multiple entities match criteria
"""
@classmethod
def get_for_update(cls, *args, **kwargs):
"""Get entity with row lock for update.
Args:
*args: Primary key values
**kwargs: Attribute conditions
Returns:
Entity instance with row lock
"""
@classmethod
def exists(cls, *args, **kwargs):
"""Check if entity exists.
Args:
*args: Primary key values or lambda expression
**kwargs: Attribute conditions
Returns:
bool: True if entity exists
"""
@classmethod
def select(cls, lambda_expr=None):
"""Select entities matching criteria.
Args:
lambda_expr: Lambda expression for filtering
Returns:
Query object with matching entities
"""
@classmethod
def select_random(cls, limit):
"""Select random entities.
Args:
limit: Number of random entities to return
Returns:
List of random entity instances
"""
def to_dict(self, only=None, exclude=None, with_collections=True,
with_lazy=True, related_objects=False):
"""Convert entity to dictionary.
Args:
only: List of attributes to include
exclude: List of attributes to exclude
with_collections: Include collection attributes
with_lazy: Include lazy-loaded attributes
related_objects: Include related entity objects
Returns:
dict: Entity data as dictionary
"""Usage Examples
Basic Database Setup
from pony.orm import *
# Create database instance
db = Database()
# Define entities
class Person(db.Entity):
name = Required(str)
age = Optional(int)
email = Optional(str, unique=True)
class Order(db.Entity):
date = Required(datetime)
person = Required(Person)
total = Required(float)
# Bind to SQLite database
db.bind('sqlite', filename='example.db')
# Generate database schema
db.generate_mapping(create_tables=True)Working with Entities
with db_session:
# Create new entities
person = Person(name="Alice", age=25, email="alice@example.com")
order = Order(date=datetime.now(), person=person, total=149.99)
# Get entity by primary key
alice = Person[1] # or Person.get(id=1)
# Get entity by unique attribute
alice = Person.get(email="alice@example.com")
# Update entity
alice.age = 26
alice.set(age=26) # alternative syntax
# Check if entity exists
if Person.exists(name="Bob"):
print("Bob exists")
# Convert to dictionary
person_data = alice.to_dict()
# Delete entity
alice.delete()Advanced Database Operations
with db_session:
# Execute raw SQL
result = db.execute("SELECT COUNT(*) FROM Person")
# Get entities with conditions
young_people = Person.select(lambda p: p.age < 30)
# Random selection
random_people = Person.select_random(5)
# Bulk operations
Person.select(lambda p: p.age < 18).delete()
# Transaction control
try:
# ... database operations
commit()
except Exception:
rollback()
raise