CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-sqlalchemy

Python SQL toolkit and Object Relational Mapper providing full power and flexibility of SQL

Overview
Eval results
Files

orm.mddocs/

Object Relational Mapping (ORM)

Declarative mapping, session management, relationship definitions, query API, and advanced ORM features including inheritance, polymorphism, and events. The ORM provides transparent object persistence using the identity map and unit of work patterns.

Capabilities

Declarative Base and Registry

Foundation classes for declarative ORM mapping.

class DeclarativeBase:
    """Modern declarative base class for SQLAlchemy 2.0+."""
    
    metadata: MetaData
    registry: registry
    
    @classmethod
    def __init_subclass__(cls, **kwargs):
        """Handle subclass registration with registry."""

def declarative_base(cls=None, bind=None, metadata=None, mapper=None, **kw):
    """
    Create declarative base class (legacy approach).
    
    Parameters:
    - cls: base class to extend (default object)
    - bind: Engine to bind metadata
    - metadata: MetaData instance to use
    - mapper: custom mapper class
    
    Returns:
    type: Declarative base class
    """

class registry:
    """Mapper registry for declarative configurations."""
    
    def __init__(self, metadata=None, **kwargs):
        """
        Create mapper registry.
        
        Parameters:
        - metadata: MetaData instance for tables
        """
    
    def configure(self):
        """Configure all pending mappers in registry."""
    
    def dispose(self):
        """Dispose all mappers and clear registry."""

Session Management

Session creation, configuration, and lifecycle management.

class Session:
    """ORM session with identity map and unit of work."""
    
    def __init__(self, bind=None, **kwargs):
        """
        Create ORM session.
        
        Parameters:
        - bind: Engine or Connection for database operations
        - autoflush: bool, auto-flush before queries (default True)
        - autocommit: bool, deprecated in 2.0
        - expire_on_commit: bool, expire objects after commit (default True)
        """
    
    def add(self, instance):
        """
        Add object instance to session.
        
        Parameters:
        - instance: mapped object to add
        """
    
    def add_all(self, instances):
        """
        Add multiple object instances to session.
        
        Parameters:
        - instances: iterable of mapped objects
        """
    
    def delete(self, instance):
        """
        Mark object instance for deletion.
        
        Parameters:
        - instance: mapped object to delete
        """
    
    def commit(self):
        """Flush pending changes and commit transaction."""
    
    def rollback(self):
        """Rollback current transaction and expire all objects."""
    
    def flush(self):
        """Flush pending changes to database without committing."""
    
    def expunge(self, instance):
        """
        Remove instance from session without deleting.
        
        Parameters:
        - instance: mapped object to remove from session
        """
    
    def expunge_all(self):
        """Remove all instances from session."""
    
    def refresh(self, instance, attribute_names=None):
        """
        Refresh object from database.
        
        Parameters:
        - instance: mapped object to refresh
        - attribute_names: specific attributes to refresh
        """
    
    def merge(self, instance):
        """
        Merge detached instance into session.
        
        Parameters:
        - instance: detached mapped object
        
        Returns:
        object: Merged persistent instance
        """
    
    def execute(self, statement, parameters=None, **kwargs):
        """
        Execute statement with ORM-level processing.
        
        Parameters:
        - statement: SQL statement or ORM query
        - parameters: bind parameters
        
        Returns:
        Result: Query results
        """
    
    def scalar(self, statement, parameters=None, **kwargs):
        """
        Execute statement and return scalar result.
        
        Parameters:
        - statement: SQL statement or ORM query
        - parameters: bind parameters
        
        Returns:
        Any: Scalar result value
        """
    
    def get(self, entity, ident):
        """
        Get object by primary key.
        
        Parameters:
        - entity: mapped class
        - ident: primary key value or tuple
        
        Returns:
        object or None: Object instance or None if not found
        """

def sessionmaker(bind=None, **kwargs):
    """
    Create Session factory.
    
    Parameters:
    - bind: Engine for database operations
    - kwargs: Session configuration options
    
    Returns:
    sessionmaker: Session factory class
    """

def scoped_session(session_factory):
    """
    Create thread-local scoped session.
    
    Parameters:
    - session_factory: sessionmaker instance
    
    Returns:
    scoped_session: Thread-local session proxy
    """

Column Mapping

Modern column mapping with type annotations and configuration.

def mapped_column(*args, **kwargs):
    """
    Define mapped column with type annotations.
    
    Parameters:
    - type_: SQLAlchemy type or Python type for annotation
    - primary_key: bool, column is primary key
    - nullable: bool, column allows NULL values
    - default: default value or callable
    - server_default: server-side default expression
    - unique: bool, column has unique constraint
    - index: bool, create index on column
    - autoincrement: bool or str, auto-increment behavior
    
    Returns:
    MappedColumn: Configured mapped column
    """

def column_property(*columns, **kwargs):
    """
    Create property from SQL expression.
    
    Parameters:
    - columns: Column or SQL expression
    - deferred: bool, defer loading until accessed
    - group: str, deferred loading group name
    
    Returns:
    ColumnProperty: SQL expression property
    """

def deferred(column, **kwargs):
    """
    Mark column for deferred loading.
    
    Parameters:
    - column: Column to defer
    - group: str, deferred loading group
    
    Returns:
    ColumnProperty: Deferred column property
    """

def synonym(name, **kwargs):
    """
    Create property synonym.
    
    Parameters:
    - name: str, name of target property
    - map_column: bool, create column mapping
    
    Returns:
    SynonymProperty: Property synonym
    """

Relationships

Relationship definition and configuration between mapped classes.

def relationship(argument, **kwargs):
    """
    Define relationship between mapped classes.
    
    Parameters:
    - argument: str or class, related class or class name
    - back_populates: str, bidirectional relationship property
    - backref: str or backref(), automatic reverse relationship
    - foreign_keys: list, foreign key columns for relationship
    - remote_side: Column or list, remote side for self-referential
    - primaryjoin: join condition for relationship
    - secondaryjoin: join condition for many-to-many
    - secondary: Table for many-to-many association
    - cascade: str, cascade behavior ('all, delete-orphan', etc.)
    - lazy: str, loading strategy ('select', 'joined', 'subquery', 'dynamic')
    - uselist: bool, relationship is one-to-many/many-to-many
    - order_by: ordering for collection relationships
    
    Returns:
    RelationshipProperty: Configured relationship
    """

def backref(name, **kwargs):
    """
    Create automatic reverse relationship.
    
    Parameters:
    - name: str, name of reverse relationship property
    - kwargs: relationship configuration for reverse side
    
    Returns:
    backref: Reverse relationship configuration
    """

def foreign(expr):
    """
    Mark expression as foreign key side of relationship.
    
    Parameters:
    - expr: Column expression
    
    Returns:
    _AnnotatedColumn: Annotated foreign column
    """

def remote(expr):
    """
    Mark expression as remote side of relationship.
    
    Parameters:
    - expr: Column expression
    
    Returns:
    _AnnotatedColumn: Annotated remote column
    """

Query and Loading Strategies

Query execution and relationship loading optimization.

def joinedload(attr):
    """
    Eagerly load relationship using JOIN.
    
    Parameters:
    - attr: relationship attribute to load
    
    Returns:
    Load: Joined loading option
    """

def selectinload(attr):
    """
    Eagerly load relationship using separate SELECT.
    
    Parameters:
    - attr: relationship attribute to load
    
    Returns:
    Load: Select-in loading option
    """

def subqueryload(attr):
    """
    Eagerly load relationship using subquery.
    
    Parameters:
    - attr: relationship attribute to load
    
    Returns:
    Load: Subquery loading option
    """

def lazyload(attr):
    """
    Lazily load relationship (default behavior).
    
    Parameters:
    - attr: relationship attribute to load
    
    Returns:
    Load: Lazy loading option
    """

def noload(attr):
    """
    Prevent automatic loading of relationship.
    
    Parameters:
    - attr: relationship attribute
    
    Returns:
    Load: No-load option
    """

def raiseload(attr):
    """
    Raise exception if relationship accessed.
    
    Parameters:
    - attr: relationship attribute
    
    Returns:
    Load: Raise-on-access option
    """

def contains_eager(attr):
    """
    Mark relationship as already loaded in query.
    
    Parameters:
    - attr: relationship attribute
    
    Returns:
    Load: Contains-eager option
    """

Object State and Inspection

Object state management and ORM inspection utilities.

def object_session(instance):
    """
    Get session associated with object instance.
    
    Parameters:
    - instance: mapped object
    
    Returns:
    Session or None: Associated session
    """

def make_transient(instance):
    """
    Make persistent object transient (unsaved).
    
    Parameters:
    - instance: persistent mapped object
    """

def make_transient_to_detached(instance):
    """
    Move transient object to detached state.
    
    Parameters:
    - instance: transient mapped object
    """

def class_mapper(class_):
    """
    Get mapper for mapped class.
    
    Parameters:
    - class_: mapped class
    
    Returns:
    Mapper: Class mapper
    """

def inspect(subject):
    """
    Inspect mapped object, class, or attribute.
    
    Parameters:
    - subject: mapped object, class, or attribute
    
    Returns:
    Inspector: Appropriate inspector object
    """

def was_deleted(object):
    """
    Check if object was deleted in current session.
    
    Parameters:
    - object: mapped object instance
    
    Returns:
    bool: True if object was deleted
    """

Advanced ORM Features

Polymorphism, aliasing, and advanced query techniques.

def aliased(element, alias=None, name=None):
    """
    Create alias for mapped class or selectable.
    
    Parameters:
    - element: mapped class or selectable
    - alias: specific alias to use
    - name: name for alias
    
    Returns:
    AliasedClass: Aliased class or selectable
    """

def with_polymorphic(base, classes, selectable=None, **kwargs):
    """
    Create polymorphic selectable for inheritance queries.
    
    Parameters:
    - base: base mapped class
    - classes: subclasses to include
    - selectable: custom selectable for polymorphic loading
    
    Returns:
    AliasedClass: Polymorphic selectable
    """

def polymorphic_union(table_map, typecolname, **kwargs):
    """
    Create UNION for polymorphic mapping.
    
    Parameters:
    - table_map: dict of discriminator->table mappings
    - typecolname: name of discriminator column
    
    Returns:
    Alias: UNION selectable for polymorphic queries
    """

class Bundle:
    """Bundle multiple columns or expressions."""
    
    def __init__(self, name, *exprs, **kwargs):
        """
        Create column bundle.
        
        Parameters:
        - name: str, bundle name
        - exprs: expressions to bundle
        - single_entity: bool, bundle represents single entity
        """

Composite and Hybrid Properties

Advanced property types for complex attribute mapping.

def composite(class_, *attrs, **kwargs):
    """
    Create composite property from multiple columns.
    
    Parameters:
    - class_: composite value class
    - attrs: column attributes for composite
    - group: str, deferred loading group
    
    Returns:
    CompositeProperty: Composite property mapping
    """

class hybrid_property:
    """Property with different SQL and Python implementations."""
    
    def __init__(self, fget, fset=None, fdel=None, expr=None):
        """
        Create hybrid property.
        
        Parameters:
        - fget: Python getter function
        - fset: Python setter function  
        - fdel: Python deleter function
        - expr: SQL expression function
        """
    
    def expression(self, func):
        """
        Decorator to define SQL expression.
        
        Parameters:
        - func: function returning SQL expression
        
        Returns:
        hybrid_property: Property with SQL expression
        """

class hybrid_method:
    """Method with different SQL and Python implementations."""
    
    def __init__(self, func, expr=None):
        """
        Create hybrid method.
        
        Parameters:
        - func: Python method implementation
        - expr: SQL expression method
        """

Session Events

Event handling for session lifecycle and object state changes.

class SessionEvents:
    """Session-level event hooks."""
    
    @staticmethod
    def after_transaction_create(session, transaction):
        """Called after transaction creation."""
    
    @staticmethod
    def after_transaction_end(session, transaction):
        """Called after transaction end."""
    
    @staticmethod
    def before_commit(session):
        """Called before transaction commit."""
    
    @staticmethod
    def after_commit(session):
        """Called after transaction commit."""
    
    @staticmethod
    def after_rollback(session):
        """Called after transaction rollback."""
    
    @staticmethod
    def before_flush(session, flush_context, instances):
        """Called before session flush."""
    
    @staticmethod
    def after_flush(session, flush_context):
        """Called after session flush."""

class AttributeEvents:
    """Attribute-level event hooks."""
    
    @staticmethod
    def set(target, value, oldvalue, initiator):
        """Called when attribute is set."""
    
    @staticmethod
    def append(target, value, initiator):
        """Called when item appended to collection."""
    
    @staticmethod
    def remove(target, value, initiator):
        """Called when item removed from collection."""

Usage Examples

Basic ORM Usage

from sqlalchemy import create_engine, String, Integer
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100))

engine = create_engine("sqlite:///example.db")
Base.metadata.create_all(engine)

# Session usage
with Session(engine) as session:
    user = User(name="John Doe", email="john@example.com")
    session.add(user)
    session.commit()
    
    # Query
    users = session.execute(
        select(User).where(User.name.like('%John%'))
    ).scalars().all()

Relationships Example

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    
    orders: Mapped[List["Order"]] = relationship(back_populates="user")

class Order(Base):
    __tablename__ = 'orders'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    amount: Mapped[Decimal] = mapped_column()
    
    user: Mapped["User"] = relationship(back_populates="orders")

# Usage with eager loading
with Session(engine) as session:
    users = session.execute(
        select(User).options(selectinload(User.orders))
    ).scalars().all()

Advanced Query Patterns

# Aliased queries
UserAlias = aliased(User)

stmt = select(User, UserAlias).join(
    UserAlias, User.manager_id == UserAlias.id
)

# Polymorphic queries
class Employee(User):
    __tablename__ = 'employees'
    id: Mapped[int] = mapped_column(ForeignKey('users.id'), primary_key=True)
    employee_number: Mapped[str] = mapped_column()

stmt = select(User).where(User.type == 'employee')
employees = session.execute(stmt).scalars().all()

Loading Strategies

Control how related objects are loaded with various loading strategies.

# Eager loading strategies
def joinedload(attr, **kwargs):
    """Load related objects using LEFT OUTER JOIN."""
    
def selectinload(attr, **kwargs):
    """Load related objects using separate SELECT IN query."""
    
def subqueryload(attr, **kwargs):
    """Load related objects using correlated subquery."""
    
def immediateload(attr, **kwargs):
    """Load related objects immediately during parent load."""

# Lazy loading control
def lazyload(attr, **kwargs):
    """Enable lazy loading for relationship (default)."""
    
def noload(attr, **kwargs):
    """Disable loading of relationship entirely."""
    
def raiseload(attr, **kwargs):
    """Raise exception when relationship is accessed."""

# Column loading strategies  
def defer(attr, **kwargs):
    """Defer loading of column until accessed."""
    
def undefer(attr, **kwargs):
    """Undefer previously deferred column."""
    
def load_only(*attrs, **kwargs):
    """Load only specified columns, defer others."""

# Relationship utilities
def contains_eager(attr, **kwargs):
    """Mark relationship as already loaded in query result."""
    
def with_polymorphic(base, classes, **kwargs):
    """Query polymorphic classes together."""

Utility Functions

Helper functions for ORM state management and inspection.

# Object state management
def make_transient(instance):
    """Make persistent object transient (remove from session)."""
    
def make_transient_to_detached(instance):
    """Make transient object detached with persistent identity."""
    
def object_session(instance):
    """Get session containing the given object instance."""

# Session utilities  
def close_all_sessions():
    """Close all sessions in current thread."""

# Mapper inspection
def class_mapper(class_):
    """Get mapper for mapped class."""
    
def object_mapper(instance):
    """Get mapper for object instance."""
    
def configure_mappers():
    """Configure all pending mappers."""
    
def clear_mappers():
    """Clear all mapper configuration."""

# Relationship configuration helpers
def foreign(expr):
    """Mark column expression as foreign key side of relationship."""
    
def remote(expr):
    """Mark column expression as remote side of relationship."""
    
def backref(name, **kwargs):
    """Create bidirectional relationship."""

Install with Tessl CLI

npx tessl i tessl/pypi-sqlalchemy

docs

async.md

core-engine.md

dialects.md

index.md

orm.md

schema.md

sql-expression.md

types.md

tile.json