flask_more_smorest.sqla

SQLAlchemy integration module.

This module provides the core SQLAlchemy integration for flask-more-smorest, including the database instance, base model, schema utilities, and migration tools.

class flask_more_smorest.sqla.BaseModel(**kwargs)[source]

Base model for all application models.

This base class provides: - Automatic UUID primary key generation - Automatic created_at and updated_at timestamps - Automatic Marshmallow schema generation via Model.Schema - Common CRUD operations (get, save, update, delete) - Lifecycle hooks (on_before_create, on_after_create, etc.)

All models should inherit from this class to get these features.

id: UUID primary key (automatically generated)

created_at: Timestamp of creation

updated_at: Timestamp of last update

Schema: Auto-generated Marshmallow schema class

Example

>>> class Article(BaseModel):
...     title: Mapped[str] = mapped_column(sa.String(200))
...     content: Mapped[str] = mapped_column(sa.Text)
...
>>> # Use auto-generated schema
>>> article_bp = CRUDBlueprint(
...     'articles', __name__,
...     model=Article,
...     schema=Article.Schema  # No need to define custom schema
... )

id: Mapped[UUID] = <sqlalchemy.orm.properties.MappedColumn object>

created_at: Mapped[datetime] = <sqlalchemy.orm.properties.MappedColumn object>

updated_at: Mapped[datetime] = <sqlalchemy.orm.properties.MappedColumn object>

__init__(**kwargs)[source]

Initialize the model.

Parameters:: **kwargs (object) – Field values to initialize the model with
Raises:: RuntimeError – If database session is not active

property is_writable: bool

Return whether the instance is writable.

BaseModel does not enforce permissions, so instances are considered writable by default. Permission-aware subclasses can override this property.

classmethod get_by(**kwargs)[source]

Get resource by field values.

Converts UUID strings to UUID objects automatically for UUID columns.

Parameters:: **kwargs (str | int | UUID | bool | None) – Field name and value pairs to filter by
Return type:: Optional[Self]
Returns:: The matching model instance, or None if not found
Raises:: TypeError – If ID is not a valid UUID string or UUID object

Example

>>> user = User.get_by(email='test@example.com')
>>> article = Article.get_by(id='123e4567-e89b-12d3-a456-426614174000')

classmethod get_by_or_404(**kwargs)[source]

Get resource by field values or raise 404.

Parameters:

**kwargs (str | int | UUID | bool | None) – Field name and value pairs to filter by

Return type:

Self

Returns:

The matching model instance

Raises:

NotFoundError – If no matching resource is found
TypeError – If ID field has invalid UUID format
ForbiddenError – If user doesn’t have read permission

Example

>>> user = User.get_by_or_404(email='test@example.com')

classmethod get(id)[source]

Get resource by ID.

Parameters:: id (UUID | str) – Resource ID (UUID or UUID string)
Return type:: Optional[Self]
Returns:: The matching model instance, or None if not found

Example

>>> user = User.get('123e4567-e89b-12d3-a456-426614174000')

classmethod get_or_404(id)[source]

Get resource by ID or raise 404.

Parameters:: id (UUID | str) – Resource ID (UUID or UUID string)
Return type:: Self
Returns:: The matching model instance
Raises:: NotFoundError – If no matching resource is found

Example

>>> user = User.get_or_404('123e4567-e89b-12d3-a456-426614174000')

classmethod check_exists(id)[source]

Check if resource exists and throw 404 otherwise.

Parameters:: id (UUID | str) – Resource ID to check
Raises:: NotFoundError – If resource doesn’t exist
Return type:: None

classmethod bypass_perms(cls)[source]

No-op context manager for base class (overridden in perms model).

Return type:: Iterator[None]

save(commit=True)[source]

Save the record: add to session and optionally commit.

Parameters:: commit (bool) – Whether to commit the transaction (default: True)
Return type:: Self
Returns:: The saved model instance (self)
Raises:: ForbiddenError – If user doesn’t have permission to create/modify

Example

>>> user = User(email='test@example.com')
>>> user.save()

commit(is_delete=False, *, is_create=None)[source]

Commit the session and call appropriate lifecycle hooks.

Parameters:

is_delete (bool) – Whether this is a delete operation (default: False)
is_create (bool | None) – Explicit flag indicating whether this commit corresponds to a creation

Return type:

None

update(commit=True, **kwargs)[source]

Update model fields using key-value pairs.

Supports updating relationships and recursively checks create permissions for nested objects.

Parameters:

commit (bool) – Whether to commit the transaction (default: True)
**kwargs (str | int | float | bool | bytes | None) – Field names and values to update

Raises:

AttributeError – If field doesn’t exist on the model
ForbiddenError – If user doesn’t have permission to modify

Return type:

None

Example

>>> user.update(email='new@example.com', is_active=False)

delete(commit=True)[source]

Delete the record from the database.

Parameters:: commit (bool) – Whether to commit the transaction (default: True)
Raises:: ForbiddenError – If user doesn’t have permission to delete
Return type:: None

Example

>>> user = User.get(user_id)
>>> user.delete()

get_clone()[source]

Return a copy of the object with a new ID.

Creates a detached copy of this instance with ID set to None, suitable for creating a duplicate record.

Return type:: Self
Returns:: A new instance with the same field values but no ID

Example

>>> original = User.get(user_id)
>>> clone = original.get_clone()
>>> clone.save()  # Creates new record

on_before_create()[source]

Hook to be called before creating the object.

Override this method to add custom logic before object creation.

Return type:: None

on_after_create()[source]

Hook to be called after creating the object.

Override this method to add custom logic after object creation.

Return type:: None

on_before_update()[source]

Hook to be called before updating the object.

Override this method to add custom logic before object updates.

Return type:: None

on_after_update()[source]

Hook to be called after updating the object.

Override this method to add custom logic after object updates.

Return type:: None

on_before_delete()[source]

Hook to be called before deleting the object.

Override this method to add custom logic before object deletion.

Return type:: None

on_after_delete()[source]

Hook to be called after deleting the object.

Override this method to add custom logic after object deletion.

Return type:: None

check_create(val)[source]

Recursively validate nested models before creating them.

Ensures nested BaseModel instances have an opportunity to perform their own permission checks (for example, BasePermsModel subclasses).

Return type:: None

__repr__()[source]

Return string representation of the model.

Return type:: str
Returns:: String in format “<ModelName id=…>”

class flask_more_smorest.sqla.BaseSchema(*args, **kwargs)[source]

Base schema for all Marshmallow schemas.

This schema extends SQLAlchemyAutoSchema with automatic view_args injection for URL parameters and adds an is_writable field for permission checking.

is_writable: Read-only boolean field indicating if current user can write to the resource

Example

>>> class ArticleSchema(BaseSchema):
...     class Meta:
...         model = Article
...         include_relationships = True

pre_load(data, **kwargs)[source]

Pre-load hook to handle UUID conversion and view_args injection.

Automatically injects URL parameters from Flask’s request.view_args into the data being loaded, allowing schemas to access route parameters.

Parameters:

data (dict[str, str | int | float | bool]) – The input data dictionary
**kwargs (Any) – Additional keyword arguments from Marshmallow

Return type:

dict[str, str | int | float | bool]

Returns:

The modified data dictionary with view_args injected

Example

Given a route /articles/<uuid:article_id> and a schema field article_id, the article_id from the URL will be automatically injected into the data if not already present.

opts: Opts = <marshmallow_sqlalchemy.schema.SQLAlchemyAutoSchemaOpts object>

instance: _ModelType | None

flask_more_smorest.sqla.create_migration(message, directory='migrations')[source]

Create a new migration file.

Automatically detects changes in the database models and generates a migration script.

Parameters:

message (str) – Description of the migration
directory (str) – Directory containing migration files (default: ‘migrations’)

Raises:

RuntimeError – If migrations directory doesn’t exist

Return type:

None

Example

>>> create_migration("Add user profile fields")

flask_more_smorest.sqla.create_model_schema(model_cls, *, db_session=None)[source]

Create a Marshmallow schema for a SQLAlchemy model class.

This function generates a schema class with sensible defaults for CRUD operations, including relationship handling and automatic timestamp field management.

Parameters:

model_cls (type) – SQLAlchemy model class to generate schema for
db_session (Any) – SQLAlchemy session to use (if None, uses model_cls’s session)

Return type:

type[BaseSchema]

Returns:

Generated BaseSchema subclass for the model

Example

>>> from flask_more_smorest.sqla import BaseModel
>>> class Article(BaseModel):
...     title: Mapped[str] = mapped_column(sa.String(200))
>>>
>>> ArticleSchema = create_model_schema(Article)
>>> schema = ArticleSchema()

flask_more_smorest.sqla.downgrade_database(revision, directory='migrations')[source]

Downgrade database to specified revision.

Reverts database migrations to the specified revision.

Parameters:

revision (str) – Target revision to downgrade to
directory (str) – Directory containing migration files (default: ‘migrations’)

Return type:

None

Example

>>> downgrade_database("-1")  # Downgrade one revision
>>> downgrade_database("ae1027a6acf")  # Downgrade to specific revision

flask_more_smorest.sqla.get_request_query_stats()[source]

Get query statistics for the current request.

Returns a dictionary with query count and total time for the current request. When called outside an application context, or when SQLALCHEMY_PERFORMANCE_MONITORING is disabled (so no query stats have been collected), this function returns zeros for both values.

Return type:: dict[str, Any]
Returns:: Dictionary with ‘query_count’ and ‘total_query_time’ keys

Example

>>> stats = get_request_query_stats()
>>> print(f"Queries: {stats['query_count']}, Time: {stats['total_query_time']:.3f}s")

flask_more_smorest.sqla.init_db(app)[source]

Initialize the database with the Flask application.

This function binds the SQLAlchemy database instance to the Flask application, making it available throughout the application context.

Parameters:: app (Flask) – Flask application instance to initialize the database with
Return type:: None

Example

>>> from flask import Flask
>>> from flask_more_smorest.sqla import init_db
>>>
>>> app = Flask(__name__)
>>> app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
>>> init_db(app)

flask_more_smorest.sqla.init_migrations(app, directory='migrations')[source]

Initialize Alembic migration environment for the application.

Creates the migrations directory structure and configuration files needed for Alembic database migrations.

Parameters:

app (Flask) – Flask application instance
directory (str) – Directory name for migration files (default: ‘migrations’)

Return type:

None

Example

>>> from flask import Flask
>>> from flask_more_smorest.sqla import init_migrations
>>>
>>> app = Flask(__name__)
>>> init_migrations(app)

flask_more_smorest.sqla.upgrade_database(revision='head', directory='migrations')[source]

Upgrade database to specified revision.

Applies database migrations up to the specified revision.

Parameters:

revision (str) – Target revision (default: ‘head’ for latest)
directory (str) – Directory containing migration files (default: ‘migrations’)

Return type:

None

Example

>>> upgrade_database()  # Upgrade to latest
>>> upgrade_database("ae1027a6acf")  # Upgrade to specific revision

Modules

`base_model`	Base model for SQLAlchemy models with automatic schema generation.
`database`	Database configuration and utilities for flask-more-smorest.
`migrations`	Migration utilities for flask-more-smorest.
`schema`	Schema generation for SQLAlchemy models.