tessl/pypi-sqlalchemy
Python SQL toolkit and Object Relational Mapper providing full power and flexibility of SQL
schema.mddocs/
Schema Definition and Reflection
Table, column, and constraint definition, database schema reflection, index management, and DDL generation for database structure management. Schema definition provides the foundation for both Core and ORM table mapping.
Capabilities
MetaData and Table Definition
Core schema container and table definition with columns and constraints.
class MetaData:
"""Container for database schema information."""
def __init__(self, bind=None, reflect=False, schema=None, **kwargs):
"""
Create MetaData container.
Parameters:
- bind: Engine for automatic table creation/reflection
- reflect: bool, automatically reflect all tables
- schema: str, default schema name
"""
def create_all(self, bind, tables=None, checkfirst=True):
"""
Create all tables in database.
Parameters:
- bind: Engine or Connection for creation
- tables: specific tables to create (all if None)
- checkfirst: bool, check if table exists before creating
"""
def drop_all(self, bind, tables=None, checkfirst=True):
"""
Drop all tables from database.
Parameters:
- bind: Engine or Connection for dropping
- tables: specific tables to drop (all if None)
- checkfirst: bool, check if table exists before dropping
"""
def reflect(self, bind, schema=None, views=False, only=None, **kwargs):
"""
Reflect database schema into MetaData.
Parameters:
- bind: Engine or Connection for reflection
- schema: str, schema name to reflect
- views: bool, reflect views as well as tables
- only: callable or list, filter for table names
"""
class Table:
"""Database table definition with columns and constraints."""
def __init__(self, name, metadata, *args, **kwargs):
"""
Define database table.
Parameters:
- name: str, table name
- metadata: MetaData container
- args: Column and Constraint objects
- schema: str, schema name
- quote: bool, quote table name
- autoload_with: Engine for automatic loading
- extend_existing: bool, extend existing table definition
- keep_existing: bool, keep existing table definition
"""
def create(self, bind, checkfirst=True):
"""
Create table in database.
Parameters:
- bind: Engine or Connection
- checkfirst: bool, check existence before creating
"""
def drop(self, bind, checkfirst=True):
"""
Drop table from database.
Parameters:
- bind: Engine or Connection
- checkfirst: bool, check existence before dropping
"""
@property
def columns(self):
"""ColumnCollection of table columns."""
@property
def c(self):
"""Shorthand for columns attribute."""
@property
def primary_key(self):
"""PrimaryKeyConstraint for table."""
@property
def foreign_keys(self):
"""Set of ForeignKey objects in table."""Column Definition
Column specification with types, constraints, and default values.
class Column:
"""Database column definition."""
def __init__(self, *args, **kwargs):
"""
Define database column.
Parameters:
- name: str, column name (optional, can be inferred)
- type_: TypeEngine, column data type
- primary_key: bool, column is primary key
- nullable: bool, column allows NULL (default True)
- default: default value or callable
- server_default: DefaultClause for server-side default
- server_onupdate: DefaultClause for server-side update
- unique: bool, column has unique constraint
- index: bool, create index on column
- autoincrement: bool or str, auto-increment behavior
- quote: bool, quote column name
- comment: str, column comment
"""
def copy(self, **kwargs):
"""
Create copy of column with modifications.
Parameters:
- kwargs: attributes to override
Returns:
Column: Copied column with changes
"""Constraints
Primary key, foreign key, unique, and check constraints for data integrity.
class PrimaryKeyConstraint:
"""Primary key constraint on one or more columns."""
def __init__(self, *columns, **kwargs):
"""
Create primary key constraint.
Parameters:
- columns: Column objects or column names
- name: str, constraint name
"""
class ForeignKeyConstraint:
"""Foreign key constraint referencing another table."""
def __init__(self, columns, refcolumns, **kwargs):
"""
Create foreign key constraint.
Parameters:
- columns: list of local column names
- refcolumns: list of referenced column names
- name: str, constraint name
- onupdate: str, ON UPDATE action ('CASCADE', 'SET NULL', etc.)
- ondelete: str, ON DELETE action
- deferrable: bool, constraint is deferrable
- initially: str, initial constraint check timing
"""
class ForeignKey:
"""Individual foreign key reference."""
def __init__(self, column, **kwargs):
"""
Create foreign key reference.
Parameters:
- column: str, referenced column (table.column format)
- onupdate: str, ON UPDATE action
- ondelete: str, ON DELETE action
- name: str, constraint name
"""
class UniqueConstraint:
"""Unique constraint on one or more columns."""
def __init__(self, *columns, **kwargs):
"""
Create unique constraint.
Parameters:
- columns: Column objects or column names
- name: str, constraint name
- deferrable: bool, constraint is deferrable
"""
class CheckConstraint:
"""Check constraint with SQL expression."""
def __init__(self, sqltext, **kwargs):
"""
Create check constraint.
Parameters:
- sqltext: str or ClauseElement, check expression
- name: str, constraint name
"""Indexes
Index definition for query performance optimization.
class Index:
"""Database index definition."""
def __init__(self, name, *expressions, **kwargs):
"""
Create database index.
Parameters:
- name: str, index name
- expressions: Column objects or SQL expressions
- unique: bool, create unique index
- quote: bool, quote index name
- postgresql_using: str, PostgreSQL index method
- postgresql_where: ClauseElement, partial index condition
- mysql_length: dict, MySQL key length specifications
"""
def create(self, bind, checkfirst=True):
"""
Create index in database.
Parameters:
- bind: Engine or Connection
- checkfirst: bool, check existence before creating
"""
def drop(self, bind, checkfirst=True):
"""
Drop index from database.
Parameters:
- bind: Engine or Connection
- checkfirst: bool, check existence before dropping
"""Default Values and Computed Columns
Column default value specification and computed column definitions.
class DefaultClause:
"""Server-side default value clause."""
def __init__(self, arg, **kwargs):
"""
Create default clause.
Parameters:
- arg: str, ClauseElement, or callable for default value
- for_update: bool, default applies to UPDATE statements
"""
class ColumnDefault:
"""Column-specific default value."""
def __init__(self, arg, **kwargs):
"""
Create column default.
Parameters:
- arg: default value, callable, or ClauseElement
- for_update: bool, default for updates
"""
class Computed:
"""Computed/generated column definition."""
def __init__(self, sqltext, **kwargs):
"""
Create computed column.
Parameters:
- sqltext: str or ClauseElement, computation expression
- persisted: bool, store computed value (vs. virtual)
"""
class Identity:
"""Identity column specification (SQL standard)."""
def __init__(self, start=None, increment=None, **kwargs):
"""
Create identity column.
Parameters:
- start: int, starting value
- increment: int, increment value
- always: bool, ALWAYS vs BY DEFAULT generation
- on_null: bool, generate on NULL insertion
"""
class Sequence:
"""Database sequence for auto-incrementing values."""
def __init__(self, name, start=None, increment=None, **kwargs):
"""
Create sequence.
Parameters:
- name: str, sequence name
- start: int, starting value
- increment: int, increment value
- minvalue: int, minimum value
- maxvalue: int, maximum value
- cycle: bool, cycle when reaching max/min
- schema: str, schema name
"""
def create(self, bind, checkfirst=True):
"""Create sequence in database."""
def drop(self, bind, checkfirst=True):
"""Drop sequence from database."""
def next_value(self):
"""
Get next value expression for queries.
Returns:
FunctionElement: Next value function call
"""DDL Generation and Execution
Custom DDL statement creation and execution.
class DDL:
"""Custom DDL statement."""
def __init__(self, statement, **kwargs):
"""
Create DDL statement.
Parameters:
- statement: str, DDL statement text
- on: str or callable, execution condition
- bind: Engine for parameter binding
"""
def execute(self, bind, target=None):
"""
Execute DDL statement.
Parameters:
- bind: Engine or Connection
- target: target object for execution context
"""
class CreateTable:
"""CREATE TABLE DDL statement."""
def __init__(self, element, **kwargs):
"""
Create table creation DDL.
Parameters:
- element: Table object to create
- include_foreign_key_constraints: bool, include FK constraints
"""
class DropTable:
"""DROP TABLE DDL statement."""
def __init__(self, element, **kwargs):
"""
Create table drop DDL.
Parameters:
- element: Table object to drop
- if_exists: bool, add IF EXISTS clause
"""Schema Utilities
Helper functions and utilities for schema operations.
def table(name, *columns, **kw):
"""
Create ad-hoc table construct.
Parameters:
- name: str, table name
- columns: Column objects
Returns:
TableClause: Table construct for queries
"""
def column(text, type_=None, **kw):
"""
Create ad-hoc column construct.
Parameters:
- text: str, column name
- type_: TypeEngine, column type
Returns:
ColumnClause: Column construct for queries
"""
def quoted_name(value, quote):
"""
Create quoted identifier.
Parameters:
- value: str, identifier value
- quote: bool, force quoting
Returns:
quoted_name: Quoted identifier object
"""Usage Examples
Basic Table Definition
from sqlalchemy import MetaData, Table, Column, Integer, String, ForeignKey
metadata = MetaData()
users = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(50), nullable=False),
Column('email', String(100), unique=True),
Column('created_at', DateTime, server_default=func.now())
)
orders = Table('orders', metadata,
Column('id', Integer, primary_key=True),
Column('user_id', Integer, ForeignKey('users.id')),
Column('amount', Numeric(10, 2)),
Column('status', String(20), default='pending')
)
# Create tables
engine = create_engine("postgresql://user:pass@localhost/db")
metadata.create_all(engine)Schema Reflection
# Reflect existing database schema
metadata = MetaData()
metadata.reflect(bind=engine, schema='public')
# Access reflected tables
users_table = metadata.tables['users']
print(f"Columns: {list(users_table.columns.keys())}")
# Partial reflection
metadata.reflect(bind=engine, only=['users', 'orders'])Constraints and Indexes
from sqlalchemy import UniqueConstraint, CheckConstraint, Index
users = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('username', String(50)),
Column('email', String(100)),
Column('age', Integer),
# Table-level constraints
UniqueConstraint('username', 'email', name='uq_user_identity'),
CheckConstraint('age >= 18', name='ck_adult_age')
)
# Separate index creation
user_email_idx = Index('idx_user_email', users.c.email)
user_email_idx.create(engine)
# Compound index
compound_idx = Index('idx_user_name_age', users.c.username, users.c.age)Advanced Column Features
from sqlalchemy import Sequence, Identity, Computed
# Identity column (SQL Server, PostgreSQL 10+)
table_with_identity = Table('documents', metadata,
Column('id', Integer, Identity(start=1000, increment=1), primary_key=True),
Column('title', String(200))
)
# Sequence-based auto-increment
user_id_seq = Sequence('user_id_seq', start=1, increment=1)
users_with_seq = Table('users', metadata,
Column('id', Integer, user_id_seq, primary_key=True),
Column('name', String(50))
)
# Computed column
products = Table('products', metadata,
Column('price', Numeric(10, 2)),
Column('tax_rate', Numeric(5, 4)),
Column('total_price', Numeric(10, 2),
Computed('price * (1 + tax_rate)', persisted=True))
)Install with Tessl CLI
npx tessl i tessl/pypi-sqlalchemy