Skip to content

Models, data and tables

This page defines the Task model, registers it, and gets its table into the database. It continues from Basics.

Define the model

MSFLib models are SQLModel classes on three base classes from msflib.models:

  • ModelBase: a table base with id, created_at and updated_at.
  • SchemaBase: the base for the Pydantic schemas that describe API input and output.
  • BaseEnum: a str enum for fixed choices.
# app/models.py
from msflib.models import BaseEnum, ModelBase, SchemaBase


class Priority(BaseEnum):
    low = "low"
    normal = "normal"
    high = "high"


class Task(ModelBase, table=True):
    title: str
    done: bool = False
    priority: Priority = Priority.normal
    owner: str


class TaskCreate(SchemaBase):
    title: str
    priority: Priority = Priority.normal


class TaskUpdate(SchemaBase):
    title: str | None = None
    done: bool | None = None
    priority: Priority | None = None


class TaskRead(SchemaBase):
    id: int
    title: str
    done: bool
    priority: Priority
    owner: str

Task is the table. SQLModel names it after the class in lower case, so the table is task, with this DDL on SQLite:

CREATE TABLE task (
    id INTEGER NOT NULL,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
    title VARCHAR NOT NULL,
    done BOOLEAN NOT NULL,
    priority VARCHAR(6) NOT NULL,
    owner VARCHAR NOT NULL,
    PRIMARY KEY (id)
)

Why four classes

MSFLib keeps the table and the schemas separate, and the actions and routers in the next pages rely on that:

  • TaskCreate and TaskUpdate hold only what a client may send. ModelAction.create builds the row from the create schema, and update applies only the fields the client actually set. Neither has owner, because owner is not client input: it comes from the request context. The action layer takes such values through an update={...} argument (see Actions and CRUD). Putting actor or scope fields on create and update schemas lets clients forge them.
  • TaskRead is the response shape. Pass it as response_model so that internal columns do not leak as you add them to the table.

All fields on TaskUpdate default to None, and update uses model_dump(exclude_unset=True), so a PATCH that sends only done changes only done.

What the bases do

SchemaBase configures Pydantic with use_enum_values=True, coerce_numbers_to_str=True and arbitrary_types_allowed=True. A schema given priority=Priority.high dumps it as the plain string "high", and TaskCreate(title=5) gives the title "5".

BaseEnum members compare equal to their string value (Priority.high == "high" is true), and str(Priority.high) is "high", not "Priority.high". The same holds after a database round trip, where the attribute may hold a plain string instead of the enum member. Compare with the enum member or with the string; you do not need .value.

Note

SQLAlchemy stores an enum column by member name, not value. With BaseEnum keep the two identical, as above, so the stored text, the API value and the Python name agree. On PostgreSQL the column becomes a native enum type, which later migrations have to alter explicitly.

Relationships

To relate a table to another one without editing either class, call Model.add_relationship(name, Target, link_model=...) after both are defined. It wraps SQLAlchemy's mapper API and is how modules attach relationships to tables you own. See Overriding models.

Register the tables

A table exists in the database only if its class has been imported before you create tables, because defining a table=True class adds it to SQLModel.metadata. Keep one place that imports every model and builds the migration config:

# app/migration.py
from msflib.db.migrations import MigrationConfig
from sqlmodel import SQLModel

from . import models  # noqa: F401  (registers the tables on SQLModel.metadata)

MIGRATION = MigrationConfig(metadata=SQLModel.metadata, script_location="alembic")

When you add MSFLib modules, import their models here too (for example from msflib.account import models as account_models). Models that relate to each other must all be imported before the first query, or SQLAlchemy cannot resolve the relationships.

Create the tables

migrate(engine, config) brings a database to the current schema, whatever state it is in, and returns what it did. Call it when the app starts:

# app/main.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI
from msflib.db.migrations import migrate
from msflib.eventbus import bind_app_emitter

from .db import engine
from .migration import MIGRATION
from .settings import settings

core = settings.scope("CORE")


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    migrate(engine, MIGRATION)
    yield


app = FastAPI(
    title=core.PROJECT_NAME,
    openapi_url=f"{core.API_V1_STR}/openapi.json",
    lifespan=lifespan,
)
app_emitter = bind_app_emitter(app)


@app.get(f"{core.API_V1_STR}/health")
def health() -> dict[str, str]:
    return {"status": "ok", "project": core.PROJECT_NAME}

Start the app and the task table exists. What migrate does depends on the database:

Database Action
SQLite (the default for development and tests) create_all: creates missing tables, never alters existing ones. No Alembic needed.
Any other database, empty create_all, then stamps Alembic at head (created). With fresh_database="upgrade" it runs the whole revision chain instead.
Alembic-tracked alembic upgrade head (upgraded).
Has tables but no Alembic version Stamps baseline_revision, then upgrades (baselined). baseline_revision must be set.
Non-SQLite, no revisions yet create_all with a warning. Nothing is dropped.

The call is atomic on PostgreSQL and SQLite, and on PostgreSQL concurrent starts take turns on an advisory lock, so several containers can run it at once. Alembic is an optional dependency (pip install "msflib[migrations]") and is imported only when it is used. script_location points at your Alembic directory (the one with env.py and versions/); it is not read on SQLite, so the placeholder above is enough while you are developing.

Once you move to PostgreSQL, generate revisions with Alembic as usual, and let alembic/env.py hand the connection over to MSFLib so the stamp and the schema change share one transaction:

# alembic/env.py
from alembic import context
from msflib.db.migrations import run_alembic_env
from sqlmodel import SQLModel

from app import models  # noqa: F401  (registers the tables)
from app.settings import settings

run_alembic_env(
    context,
    SQLModel.metadata,
    database_url=settings.scope("CORE").SQLALCHEMY_DATABASE_URI,
)

alembic revision --autogenerate and alembic upgrade still work from the command line. run_alembic_env supports only Alembic's default alembic_version table.

An initial-data command

For deployments, a separate command that migrates and seeds is safer than doing it in every worker's startup. run_initial_data_cli builds one from your engine, migration config and a seed function:

# app/initial_data.py
import sys

from msflib.db.cli import run_initial_data_cli
from sqlalchemy.engine import Engine

from .db import engine
from .migration import MIGRATION
from .settings import settings


def seed_initial_data(engine: Engine) -> None:
    """Idempotent seeding goes here. Nothing to seed yet."""


def main(argv=None) -> int:
    return run_initial_data_cli(
        engine=engine,
        config=MIGRATION,
        seed=seed_initial_data,
        environment=settings.scope("CORE").ENVIRONMENT,
        argv=argv,
    )


if __name__ == "__main__":
    sys.exit(main())

python -m app.initial_data --check   # report only; exit 0 = nothing to do, 3 = changes pending
python -m app.initial_data           # migrate, then seed; safe to run on every deploy

On an empty SQLite database --check prints:

Database state: empty
migrate would: create_all
Stored revision: (none)
Head revision: (n/a)
Pending revisions: (none)
Differs from the models: missing table task

and exits with 3. After a plain run it reports Nothing to do and exits 0. Without Alembic (SQLite) the report can only detect missing tables, not changed columns, so an empty report there does not prove the schema matches the models.

--reset drops every table and rebuilds them, for development only. It is refused when ENVIRONMENT (CORE__ENVIRONMENT in a composed host) is production or prod, and --yes does not override that. MSFLib treats every database as sensitive except a SQLite file under ./tmp/ or /tmp/, so on the tasks.db used here a reset needs --yes (or typing RESET at the prompt).

The destructive alternative

msflib.db.init_db.init_db(engine, metadata, create_tables=True) drops every table in metadata and recreates them, then optionally runs seeders. Use it in throwaway scripts and tests, not in anything that touches data you want to keep. In tests, the usual pattern is SQLModel.metadata.create_all(engine) on a fresh in-memory engine, as shown on the last page.

Troubleshooting

no such table: task. The tables were never created, or the model was not imported before migrate ran. Check that migration.py imports models and that the app (or initial_data) imports migration.py.

A new column is missing after you edit a model. On SQLite, create_all never alters existing tables, and migrate does not either. Delete the dev database, or run python -m app.initial_data --reset --yes. On PostgreSQL, write an Alembic revision.

MigrationError: The database has tables but no Alembic version. The database was built with create_all earlier and now has Alembic revisions. Set MigrationConfig.baseline_revision to the revision it matches.

Refusing --reset. Either ENVIRONMENT (CORE__ENVIRONMENT in a composed host) is production, or the database is not a throwaway temp file and --yes was not passed.

Next: Actions and CRUD.