Skip to content

Writing custom code with MSFLib

This guide builds a small host app with one feature of its own, personal notes, next to the MSFLib auth and account modules. It uses the same pieces every MSFLib feature uses: a settings namespace, a table, an action, an authenticated router and a test. Read How to extend MSFLib first if you want the overview.

The app has this layout, and each section below is one file. All of it was run as shown, with an in-memory SQLite database.

myapp/
    __init__.py
    settings.py
    models.py
    actions.py
    deps.py
    routers/
        __init__.py
        notes.py
    main.py

1. Settings: a namespace of your own

A settings class for your feature is a ModuleSettingsBase with a namespace. Mix it into the Settings class next to the modules you use.

# myapp/settings.py
from typing import ClassVar

from msflib.account.config import AccountSettings
from msflib.auth.config import AuthSettings
from msflib.core.config import CoreSettings, ModuleSettingsBase, SettingsBase


class NotesSettings(ModuleSettingsBase):
    namespace: ClassVar[str] = "NOTES"

    MAX_NOTES_PER_ACCOUNT: int = 100


class Settings(NotesSettings, AccountSettings, AuthSettings, CoreSettings, SettingsBase):
    SECRET_KEY: str = "change-me"
    SQLITE_DATABASE_URI: str = "sqlite://"
    USERS_OPEN_REGISTRATION: bool = True


settings = Settings()

Read your values with settings.scope("NOTES").MAX_NOTES_PER_ACCOUNT. Setting the environment variable MAX_NOTES_PER_ACCOUNT=2 changes it. Other code, including MSFLib modules, can be given the same settings object.

This guide uses the subclassed settings style: the auth router reads a top-level SECRET_KEY.

Known issue (#287)

With composed (nested) settings, POST /login and the password-recovery endpoints of the auth router fail with AttributeError: ... no attribute 'SECRET_KEY'. Keep to the subclassed style shown here when you mount the auth router.

2. Models: tables and schemas

The account module does not ship a table for accounts. You define Account and Profile from the module's bases, which is also where you add your own columns (see Overriding models). Your own tables derive from ModelBase, which adds id, created_at and updated_at, and from SchemaBase.

# myapp/models.py
from typing import Optional

from msflib.account.models import AccountBase, ProfileBase
from msflib.models import ModelBase, SchemaBase
from sqlmodel import Field, Relationship


class Account(AccountBase, table=True):
    profile: Optional["Profile"] = Relationship(
        sa_relationship_kwargs={"uselist": False}, back_populates="account"
    )


class Profile(ProfileBase, table=True):
    account: Account | None = Relationship(back_populates="profile")


class NoteBase(SchemaBase):
    title: str
    body: str = ""


class Note(NoteBase, ModelBase, table=True):
    account_id: int = Field(foreign_key="account.id", index=True)


class NoteCreate(NoteBase):
    pass


class NoteUpdate(SchemaBase):
    title: str | None = None
    body: str | None = None


class NoteRead(NoteBase, ModelBase):
    id: int
    account_id: int

Keep *Create and *Update schemas to what a client may send. Values that come from the request context, such as account_id here, are not schema fields; they are passed to the action separately (next section).

The relationship annotation must be Optional["Profile"] rather than "Profile | None", because SQLAlchemy resolves the string by class name.

3. Actions: your rules

An action is the service layer for one model. ModelAction[Note, NoteCreate, NoteUpdate] already gives you get, get_by_all, get_multi_by_all, create, update, delete and more. Subclass it to add queries and rules. The details are in Subclassing actions.

# myapp/actions.py
from typing import Any

from fastapi import HTTPException
from msflib.account.actions import AccountAction, ProfileAction
from msflib.account.models import AccountCreate, AccountUpdate, ProfileCreate, ProfileUpdate
from msflib.actions import ModelAction
from sqlmodel import Session, func, select

from .models import Account, Note, NoteCreate, NoteUpdate, Profile
from .settings import settings


class NoteAction(ModelAction[Note, NoteCreate, NoteUpdate]):
    def __init__(self, *, max_per_account: int) -> None:
        self.max_per_account = max_per_account

    def list_for_account(self, session: Session, *, account_id: int, limit: int = 50) -> list[Note]:
        return self.get_multi_by_all(session, account_id=account_id, limit=limit)

    def create(
        self,
        session: Session,
        *,
        data: NoteCreate,
        update: dict[str, Any] | None = None,
        decorator=None,
        commit: bool = True,
    ) -> Note:
        account_id = (update or {}).get("account_id")
        count = session.exec(select(func.count()).where(Note.account_id == account_id)).one()
        if count >= self.max_per_account:
            raise HTTPException(status_code=409, detail="Note limit reached")
        return super().create(
            session, data=data, update=update, decorator=decorator, commit=commit
        )


pa = ProfileAction[Profile, ProfileCreate, ProfileUpdate]()
aa = AccountAction[Account, AccountCreate, AccountUpdate](settings=settings, profile_action=pa)
na = NoteAction(max_per_account=settings.scope("NOTES").MAX_NOTES_PER_ACCOUNT)

pa and aa are the account module's actions bound to your table classes. They are passed to the account router below.

4. Dependencies: session, keystore and the current account

Create the engine and the shared dependencies once. get_session_factory returns a dependency that yields a Session. get_keystore_factory returns a dependency for the token store, which holds revoked tokens (see Authentication). get_account_dependencies returns a namespace with get_current_account, get_current_active_account, RoleCheck and more.

# myapp/deps.py
from msflib.api.deps import get_keystore_factory, get_session_factory
from msflib.auth.deps import get_account_dependencies
from sqlalchemy.pool import StaticPool
from sqlmodel import create_engine

from .models import Account
from .settings import settings

engine = create_engine(
    settings.SQLITE_DATABASE_URI,
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)
get_session = get_session_factory(engine)
get_keystore = get_keystore_factory()

auth_deps = get_account_dependencies(
    AccountModel=Account,
    oauth_token_url=f"{settings.scope('CORE').API_V1_STR}/login",
    secret_key=settings.scope("CORE").SECRET_KEY,
    session_dep=get_session,
    keystore_dep=get_keystore,
)

The StaticPool engine is for a single-process development database that lives in memory. Use a real database URL in production. For one dependency namespace per deps.py, see Dependency injection.

5. A router of your own

Your endpoints are ordinary FastAPI. Use the shared dependencies, and scope every query to the caller. A dependency that loads the note only if the caller owns it keeps that rule in one place.

# myapp/routers/notes.py
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session

from ..actions import na
from ..deps import auth_deps, get_session
from ..models import Note, NoteCreate, NoteRead, NoteUpdate

router = APIRouter(prefix="/notes", tags=["notes"])


@router.get("/", response_model=list[NoteRead])
def list_notes(
    session: Session = Depends(get_session),
    account=Depends(auth_deps.get_current_active_account),
):
    return na.list_for_account(session, account_id=account.id)


@router.post("/", response_model=NoteRead, status_code=201)
def create_note(
    data: NoteCreate,
    session: Session = Depends(get_session),
    account=Depends(auth_deps.get_current_active_account),
):
    return na.create(session, data=data, update={"account_id": account.id})


def get_own_note(
    note_id: int,
    session: Session = Depends(get_session),
    account=Depends(auth_deps.get_current_active_account),
) -> Note:
    note = na.get_by_all(session, id=note_id, account_id=account.id)
    if note is None:
        raise HTTPException(status_code=404, detail="Note not found")
    return note


@router.put("/{note_id}", response_model=NoteRead)
def update_note(
    data: NoteUpdate,
    note: Note = Depends(get_own_note),
    session: Session = Depends(get_session),
):
    return na.update(session, model=note, data=data)


@router.delete("/{note_id}", status_code=204)
def delete_note(note: Note = Depends(get_own_note), session: Session = Depends(get_session)):
    na.delete(session, note.id)

update={"account_id": ...} on create overlays values onto the payload before the row is built, and update on update does the same for changes. This is the supported way to set owner and scope fields that the client must not control.

6. The app

Create tables, bind an event emitter to the app, and include the module routers and your own. The module routers are factories: you pass your table types, dependencies and settings, and you choose the prefix.

# myapp/main.py
from fastapi import FastAPI
from msflib.account.models import AccountRead
from msflib.account.router import account_router
from msflib.auth.router import router as auth_router
from msflib.eventbus import bind_app_emitter
from sqlmodel import SQLModel

from .actions import aa, pa
from .deps import auth_deps, engine, get_keystore, get_session
from .models import Account, Profile
from .routers import notes
from .settings import settings

SQLModel.metadata.create_all(engine)

app = FastAPI(title=settings.scope("CORE").PROJECT_NAME)
app_emitter = bind_app_emitter(app)
api_v1 = settings.scope("CORE").API_V1_STR

app.include_router(
    auth_router(
        get_session=get_session,
        get_keystore=get_keystore,
        get_current_account=auth_deps.get_current_account,
        account_type=Account,
        account_read_type=AccountRead,
        settings=settings,
        prefix="",
    ),
    prefix=api_v1,
)
app.include_router(
    account_router(
        get_session=get_session,
        get_current_account=auth_deps.get_current_account,
        settings=settings,
        account_type=Account,
        profile_type=Profile,
        account_action=aa,
        profile_action=pa,
        prefix="/accounts",
    ),
    prefix=api_v1,
)
app.include_router(notes.router, prefix=api_v1)

create_all is for development. For a deployed database, use migrations (see msflib.db.migrations on the core page). bind_app_emitter(app) gives the app its own event bus; actions emit lifecycle events on it during requests and in BackgroundTasks. Register listeners with @app_emitter.on("event-name"); code outside a request (scripts, threads) needs with use_app_emitter(app):, and a worker process, which has no app, registers its listeners on AppEmitter(get_emitter()). See Subclassing actions.

Try it with the test client. Open registration is enabled in the settings above, so a client can create an account, log in, and use the notes endpoints with the returned bearer token.

from fastapi.testclient import TestClient

from myapp.main import app

client = TestClient(app)
client.post("/api/v1/accounts/open", json={"email": "a@example.com", "password": "pw"})

login = client.post("/api/v1/login", data={"username": "a@example.com", "password": "pw"})
headers = {"Authorization": f"Bearer {login.json()['access_token']}"}

created = client.post("/api/v1/notes/", json={"title": "First"}, headers=headers)
assert created.status_code == 201
assert client.get("/api/v1/notes/", headers=headers).json()[0]["title"] == "First"
assert client.get("/api/v1/notes/").status_code == 401

With MAX_NOTES_PER_ACCOUNT=2 set in the environment, the third POST returns 409 with Note limit reached.

Testing your app

Every module takes its session and keystore through the callables you passed in, so tests swap them with FastAPI's dependency_overrides. Override the same function objects (get_session, get_keystore) that you passed to the factories.

from fastapi.testclient import TestClient
from msflib.core.store import MapStore
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, SQLModel, create_engine

from myapp.deps import get_keystore, get_session
from myapp.main import app

test_engine = create_engine(
    "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)
SQLModel.metadata.create_all(test_engine)


def override_session():
    with Session(test_engine) as session:
        yield session


app.dependency_overrides[get_session] = override_session
store = MapStore()
app.dependency_overrides[get_keystore] = lambda: store

client = TestClient(app)

Create the MapStore once, outside the lambda. If the override builds a new store per request, a token revoked by DELETE /logout is still accepted on the next call.

Where to go next