tessl/pypi-flask-sqlalchemy

Add SQLAlchemy support to your Flask application.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Session Management

Name: tessl/pypi-flask-sqlalchemy
Author: tessl

Session management with Flask application context integration, bind key resolution, and automatic session cleanup. Core database operations and transaction handling.

Capabilities

Session Class

Enhanced SQLAlchemy session with Flask application context integration and automatic bind selection.

class Session(sa_orm.Session):
    """
    SQLAlchemy Session class with Flask-specific bind key resolution.
    
    Automatically selects the appropriate database engine based on
    the bind key associated with the metadata of queried models/tables.
    """
    
    def __init__(self, db: SQLAlchemy, **kwargs: Any) -> None:
        """
        Initialize session with reference to SQLAlchemy extension.
        
        Parameters:
        - db: SQLAlchemy extension instance
        - kwargs: Additional session arguments
        """
    
    def get_bind(
        self,
        mapper: Any | None = None,
        clause: Any | None = None,
        bind: sa.engine.Engine | sa.engine.Connection | None = None,
        **kwargs: Any,
    ) -> sa.engine.Engine | sa.engine.Connection:
        """
        Select database engine based on bind key of queried models/tables.
        
        Parameters:
        - mapper: Model mapper or class
        - clause: SQL clause or table
        - bind: Explicit bind to use (overrides automatic selection)
        - kwargs: Additional arguments
        
        Returns:
        Database engine or connection for the query
        """

Session Access

The extension provides a scoped session accessible through the session property.

@property
def session(self) -> sa_orm.scoped_session[Session]:
    """
    Scoped session factory for database operations.
    
    Sessions are scoped to Flask application context and automatically
    removed when the context exits. Requires active app context.
    """

Query Convenience Methods

Extension methods that combine session operations with Flask error handling.

def get_or_404(
    self,
    entity: type[Any],
    ident: Any,
    *,
    description: str | None = None,
    **kwargs: Any,
) -> Any:
    """
    Get entity by primary key or abort with 404 Not Found.
    
    Parameters:
    - entity: Model class to query
    - ident: Primary key value
    - description: Custom error message
    - kwargs: Additional arguments for session.get()
    
    Returns:
    Model instance
    
    Raises:
    404 error if entity not found
    """

def first_or_404(
    self, statement: sa.sql.Select[Any], *, description: str | None = None
) -> Any:
    """
    Execute select statement and return first result or abort with 404.
    
    Parameters:
    - statement: SQLAlchemy select statement
    - description: Custom error message
    
    Returns:
    First result from query
    
    Raises:
    404 error if no results found
    """

def one_or_404(
    self, statement: sa.sql.Select[Any], *, description: str | None = None
) -> Any:
    """
    Execute select statement expecting exactly one result or abort with 404.
    
    Parameters:
    - statement: SQLAlchemy select statement  
    - description: Custom error message
    
    Returns:
    Single result from query
    
    Raises:
    404 error if no results or multiple results found
    """

Session Pagination

Built-in pagination support for select statements.

def paginate(
    self,
    select: sa.sql.Select[Any],
    *,
    page: int | None = None,
    per_page: int | None = None,
    max_per_page: int | None = None,
    error_out: bool = True,
    count: bool = True,
) -> Pagination:
    """
    Apply pagination to a select statement.
    
    Parameters:
    - select: SQLAlchemy select statement to paginate
    - page: Current page number (from request args if None) 
    - per_page: Items per page (from request args if None)
    - max_per_page: Maximum allowed per_page value
    - error_out: Abort with 404 on invalid page parameters
    - count: Calculate total item count
    
    Returns:
    Pagination object with results and metadata
    """

Utility Functions

Helper functions for session and bind management.

def _clause_to_engine(
    clause: sa.ClauseElement | None,
    engines: Mapping[str | None, sa.engine.Engine],
) -> sa.engine.Engine | None:
    """
    Get engine from table metadata bind key.
    
    Parameters:
    - clause: SQL clause containing table reference
    - engines: Available engines by bind key
    
    Returns:
    Engine for the clause's bind key or None
    """

def _app_ctx_id() -> int:
    """
    Get Flask application context ID for session scoping.
    
    Returns:
    Unique identifier for current app context
    """

Usage Examples

Basic Session Usage

# Session is automatically available and scoped to app context
with app.app_context():
    # Add new record
    user = User(username='john', email='john@example.com')
    db.session.add(user)
    db.session.commit()
    
    # Query records
    users = db.session.execute(db.select(User)).scalars().all()

Using Query Convenience Methods

@app.route('/user/<int:user_id>')
def get_user(user_id):
    # Automatically returns 404 if user not found
    user = db.get_or_404(User, user_id)
    return f"User: {user.username}"

@app.route('/users/first')
def first_user():
    # Get first user or 404 if none exist
    user = db.first_or_404(db.select(User).order_by(User.id))
    return f"First user: {user.username}"

@app.route('/user/unique/<username>')  
def unique_user(username):
    # Expect exactly one match or 404
    user = db.one_or_404(db.select(User).where(User.username == username))
    return f"User: {user.username}"

Multiple Database Sessions

# Session automatically selects correct engine based on model bind key
class User(db.Model):
    __bind_key__ = 'users'
    id = db.Column(db.Integer, primary_key=True)

class Log(db.Model):
    __bind_key__ = 'logs'
    id = db.Column(db.Integer, primary_key=True)

with app.app_context():
    # These use different databases automatically
    db.session.add(User(name='john'))  # Uses 'users' database
    db.session.add(Log(message='test'))  # Uses 'logs' database
    db.session.commit()

Session Pagination

@app.route('/users')
def list_users():
    # Automatically handles page and per_page from request args
    pagination = db.paginate(
        db.select(User).order_by(User.username),
        per_page=20,
        max_per_page=100
    )
    
    return {
        'users': [u.username for u in pagination.items],
        'page': pagination.page,
        'pages': pagination.pages,
        'total': pagination.total
    }

Manual Transaction Control

with app.app_context():
    try:
        user = User(username='john')
        db.session.add(user)
        
        # Explicit flush to get ID without committing
        db.session.flush()
        user_id = user.id
        
        # More operations...
        db.session.commit()
    except Exception:
        db.session.rollback()
        raise

Session Configuration

# Custom session options during extension initialization
db = SQLAlchemy(
    app,
    session_options={
        'class_': CustomSession,  # Use custom session class
        'expire_on_commit': False,  # Don't expire objects after commit
        'autoflush': False,  # Manual flush control
    }
)