Actions and CRUD¶

An action is the service layer for one model: the object that reads and writes its rows. ModelAction gives you create, read, update and delete without writing queries, plus filtered and paged reads and lifecycle events. Routers call actions; so do seed scripts, background jobs and tests. This page continues from Models, data and tables.

Create an action¶

ModelAction is generic over three types: the table, the create schema and the update schema. Subclass it with concrete types, which gives you a place for your own queries:

# app/actions.py
from msflib.actions import ModelAction
from sqlmodel import Session

from .models import Task, TaskCreate, TaskUpdate


class TaskAction(ModelAction[Task, TaskCreate, TaskUpdate]):
    def list_open(self, session: Session, *, owner: str) -> list[Task]:
        return self.get_multi_by_expressions(
            session,
            Task.owner == owner,
            Task.done == False,  # noqa: E712
            order_by=Task.id,
        )


task_action = TaskAction()

Create one instance and share it; actions hold no per-request state. If you have nothing to add, skip the subclass: ModelAction[Task, TaskCreate, TaskUpdate]() works, and so does msflib.actions.action(Task, TaskCreate, TaskUpdate), which returns a cached instance per type triple. Instantiating a bare ModelAction() without type arguments raises TypeError: ModelAction must be instantiated with concrete generic types. the first time it needs the model.

Every method takes a SQLModel Session as its first argument. The action never opens or closes sessions; you do, or a FastAPI dependency does (see the next page).

Create, read, update, delete¶

This script runs against an in-memory SQLite database:

from msflib.db.sqlite import enable_savepoints
from sqlmodel import Session, SQLModel, create_engine

from app import models  # noqa: F401
from app.actions import task_action
from app.models import TaskCreate, TaskUpdate

engine = enable_savepoints(create_engine("sqlite://"))
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    task = task_action.create(
        session,
        data=TaskCreate(title="write docs", priority="high"),
        update={"owner": "ann"},
    )
    print(task.id, task.title, task.owner, task.done)
    # 1 write docs ann False

    mine = task_action.get_by_all(session, id=task.id, owner="ann")
    theirs = task_action.get_by_all(session, id=task.id, owner="bob")
    print(mine is task, theirs)
    # True None

    task = task_action.update(session, model=task, data=TaskUpdate(done=True))
    print(task.done)
    # True

    deleted = task_action.delete(session, task.id)
    print(deleted.title, task_action.get(session, task.id))
    # write docs None

Points to notice:

create returns the committed row, refreshed, so id and created_at are filled in.
update takes the row you loaded, not an id. It applies only the fields the update schema has set, so TaskUpdate(done=True) leaves the title alone.
delete takes an id and returns the deleted row, or None if there was no such row.
Reads return None for no match. Turning that into a 404 is the router's job.

Methods¶

Group	Methods
One row	`get(session, id)`, `get_by_all(session, filters)`, `get_by_any(session, filters)`, `get_by_expressions(session, *exprs)`
Many rows	`get_multi`, `get_multi_by_all`, `get_multi_by_any`, `get_multi_by_expressions`, each with `offset=0` and `limit=100`
Create	`create(session, data=, update=, decorator=, commit=)`, `create_multi(session, data=[...], ...)`
Update	`update(session, model=, data=, update=, decorator=, commit=)`
Delete	`delete(session, id)`, `delete_by_all`, `delete_by_any`, `delete_by_expressions`, `delete_multi_by_all`, `delete_multi_by_any`, `delete_multi_by_expressions`
Test data	`random(overrides)`, `random_multi(count=, overrides)`, `create_random(session, count=, **overrides)`

*_by_all filters are keyword arguments that must all match (owner="ann", done=False); *_by_any matches if any does. *_by_expressions takes SQLAlchemy expressions such as Task.id > 10. The get_multi* methods return at most 100 rows unless you pass limit=; pass limit=None for no limit. Pass order_by= to get_multi_by_expressions when you page, otherwise the row order between calls is not defined.

Known issue (#276)

delete_by_expressions is annotated -> list[ModelType] but returns a single row or None. Type checkers will accept list operations on the result; do not rely on the annotation.

Scoping: values the client does not supply¶

The update={...} argument on create and update merges extra values over what the schema produced. Use it for everything that comes from the request context rather than the request body: the owner, the workspace, the tenant. That is why TaskCreate has no owner:

task_action.create(session, data=payload, update={"owner": current_owner})

On create, keys in update that are not model fields are dropped, and so are schema fields the model does not have. For reads, scope with the filters. Scope every read and write by the owner (or workspace, or tenant) the same way, so a row id alone never selects data:

mine = task_action.get_by_all(session, id=task_id, owner=current_owner)

decorator= takes a function called with the model before it is saved, for last-minute changes that are not plain field values. In real apps the identity values come from msflib-auth dependencies, and the scope dimensions (tenant, workspace, account) from get_scope_dependencies.

Transactions¶

By default each call commits. Pass commit=False to flush instead and leave the commit to you, so several writes succeed or fail together:

with Session(engine) as session:
    task_action.create(session, data=TaskCreate(title="a"), update={"owner": "ann"}, commit=False)
    task_action.create(session, data=TaskCreate(title="b"), update={"owner": "ann"}, commit=False)
    session.commit()

A rollback discards both. On SQLite this needs the engine wrapped with enable_savepoints, as in Basics. Rows created with commit=False already have their id after the flush.

Lifecycle events¶

Every write also emits events, so other code can react without the router knowing about it. For Task the names are task-create-pre-commit, task-update-pre-commit, task-delete-pre-commit and their -post-commit counterparts, and generic model-... versions of each. Pre-commit listeners run after the flush, inside the transaction, and see the live row and the session; an exception in one aborts the write. Post-commit listeners run once the outermost commit succeeds, receive a plain snapshot of the row, and cannot affect the write.

@app_emitter.on("task-create-post-commit")
def announce(task, options):
    print("created task", task.id)

Listeners are registered on the app emitter you bound in main.py (app_emitter = bind_app_emitter(app)). The app emitter is active during requests, including BackgroundTasks and tasks started from the request. Only writes made through a ModelAction emit events; session.add(...) does not. See Event bus and, for running code inside every write, on_model_write on the core page.

Test data¶

task_action.random() builds a create schema with random values for the required fields, and random(title="fixed") pins some. Because owner is context rather than input, build the schema and create it yourself so you can pass update:

data = task_action.random(title="fixed")
task = task_action.create(session, data=data, update={"owner": "ann"})

create_random(session, count=2) creates rows in one call.

Known issue (#276)

create_random has no way to pass update, so it only suits models whose columns are all in the create schema. Calling it for Task fails with an IntegrityError on the missing owner. Build the schema with random() and call create yourself, as above.

Troubleshooting¶

TypeError: ModelAction must be instantiated with concrete generic types. Subclass ModelAction[Model, Create, Update] or index it when you instantiate.

IntegrityError: NOT NULL constraint failed: task.owner. A column that is not in the create schema was not supplied through update={...}.

A listener never runs. The write bypassed ModelAction, or the listener is on a different emitter. In scripts, threads and lifespan code, wrap the code in with use_app_emitter(app): (from msflib.eventbus); a worker process has no app, so register its listeners on AppEmitter(get_emitter()). Outside a request, events go to the default emitter, which the app emitter's listeners never see.

Next: Endpoints and routers.