One application, step by step

01

An application with plugins

The plugins directory is derived from the file, not from the working directory. The tables plugins declare are created by the kernel at create_app (get_declared_tables()); the first accounts are created by keepup-auth. The bus channel is needed in the fourth example.

import os

from keepup import KeepupSettings, create_app

HERE = os.path.dirname(os.path.abspath(__file__))
PLUGINS_DIR = os.path.join(HERE, "plugins")

app = create_app(KeepupSettings(
    title="Notes",
    project_name="notes",
    plugins_dir=PLUGINS_DIR,
    notification_channel="notes_events",
    static_mounts=(),
))

{
  "plugins": [
    {"id": "notes", "name": "Notes", "enabled": true, "priority": 100, "config": {}}
  ],
  "roles": [
    {"name": "ADMIN", "modules": [], "plugins": ["notes"]},
    {"name": "CLIENT", "modules": [], "plugins": ["notes"]}
  ]
}

02

A plugin with its own table and routes

The table is declared once and created at every start if it is missing. Routes are dictionaries; the params mask checks and converts parameters before the handler is called. Database calls are synchronous, so they go to a thread and do not hold the event loop.

import asyncio
from typing import Any, Dict, List

from fastapi import HTTPException
from sqlalchemy import Column, DateTime, Integer, Text
from sqlalchemy import text as sql_text

from keepup_db import tables
from keepup_db import DatabaseManagerV2
from keepup.plugins.base import BasePlugin

MAX_NOTE_LENGTH = 2000

notes = tables.table(
    "notes",
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("owner_id", Integer, nullable=False),
    Column("text", Text, nullable=False),
    Column("created_at", DateTime, server_default=sql_text("CURRENT_TIMESTAMP")),
    tables.foreign_key("owner_id", "users", ("id",), ondelete="CASCADE"),
)


class NotesPlugin(BasePlugin):          # id "notes" -> NotesPlugin
    def __init__(self, config: Dict = None):
        super().__init__("notes", "Notes", config or {})

    async def initialize(self) -> bool:
        tables.ensure_tables(notes)   # a plugin also declares its tables: get_declared_tables()
        return True

    def get_handlers(self) -> Dict[str, Any]:
        return {"count_notes": self.count_notes}   # for other plugins

    def get_api_routes(self) -> List[Dict]:
        return [
            {"path": "/api/notes", "methods": ["GET"], "handler": self.list_notes,
             "params": {"limit": {"type": "int", "min": 1, "max": 100, "required": False}}},
            {"path": "/api/notes", "methods": ["POST"], "handler": self.add_note},
            {"path": "/api/notes/{note_id}", "methods": ["DELETE"], "handler": self.delete_note,
             "params": {"note_id": {"type": "int"}}},
        ]

    async def list_notes(self, current_user: dict, limit: int = 20):
        rows = await asyncio.to_thread(
            DatabaseManagerV2.execute,
            "SELECT id, text, created_at FROM notes WHERE owner_id = :owner "
            "ORDER BY id DESC LIMIT :limit",
            {"owner": current_user["id"], "limit": limit})
        return {"notes": rows}

    async def add_note(self, current_user: dict, request: dict = None):
        text = str((request or {}).get("text", "")).strip()
        if not text or len(text) > MAX_NOTE_LENGTH:
            raise HTTPException(status_code=400, detail="text is empty or too long")
        await asyncio.to_thread(
            DatabaseManagerV2.execute_commit,
            "INSERT INTO notes (owner_id, text) VALUES (:owner, :text)",
            {"owner": current_user["id"], "text": text})
        return {"ok": True}

    async def delete_note(self, current_user: dict, note_id: int, request: dict = None):
        removed = await asyncio.to_thread(
            DatabaseManagerV2.execute_commit,
            "DELETE FROM notes WHERE id = :id AND owner_id = :owner",
            {"id": note_id, "owner": current_user["id"]})
        if not removed:
            raise HTTPException(status_code=404, detail="no such note")
        return {"ok": True}

    def count_notes(self, owner_id: int) -> int:
        row = DatabaseManagerV2.execute_one(
            "SELECT COUNT(*) AS n FROM notes WHERE owner_id = :owner", {"owner": owner_id})
        return row["n"]

Trying it from the command line — after signing into the panel, or with a token:

curl -H "Authorization: Bearer $TOKEN" "http://localhost:8000/api/notes?limit=5"
curl -H "Authorization: Bearer $TOKEN" "http://localhost:8000/api/notes?limit=500"   # 400: outside the mask

03

Background work in one replica

The loop starts in every replica, but the purge runs only in the one that took the lock; the others skip the pass. The lock is renewed while the work goes on, so a long pass is not taken over.

from datetime import datetime, timedelta

from keepup.locks import with_distributed_lock

PURGE_EVERY_SECONDS = 3600
KEEP_NOTES_DAYS = 90


@with_distributed_lock("notes_purge", timeout=60)
async def purge_old_notes():
    cutoff = datetime.utcnow() - timedelta(days=KEEP_NOTES_DAYS)
    await asyncio.to_thread(
        DatabaseManagerV2.execute_commit,
        "DELETE FROM notes WHERE created_at < :cutoff", {"cutoff": cutoff})


class NotesPlugin(BasePlugin):
    async def initialize(self) -> bool:
        tables.ensure_tables(notes)   # a plugin also declares its tables: get_declared_tables()
        self._purger = asyncio.create_task(self._purge_forever())
        return True

    async def _purge_forever(self):
        while True:
            await asyncio.sleep(PURGE_EVERY_SECONDS)
            await purge_old_notes()      # None when another replica purges

    async def cleanup(self):
        self._purger.cancel()

04

Events, the bus and a live socket

A user's socket is held by one replica, while the note may have been added through another. deliver() hands the envelope to every replica, this one included; each writes to the sockets it holds itself. The envelope names the note rather than carrying it: delivery is best-effort, and the client re-reads the list itself.

from fastapi import WebSocketDisconnect

from keepup import notification_bus
from keepup_audit.events import emit_event


class NotesPlugin(BasePlugin):
    def __init__(self, config: Dict = None):
        super().__init__("notes", "Notes", config or {})
        self.sockets: Dict[int, set] = {}   # user id -> sockets in THIS replica

    async def initialize(self) -> bool:
        tables.ensure_tables(notes)   # a plugin also declares its tables: get_declared_tables()
        bus = notification_bus.get_notification_bus()
        if bus is not None:
            bus.subscribe(self.on_envelope)
        return True

    def get_websocket_routes(self) -> List[Dict]:
        # Sign-in is off for sockets by default -- switch it on.
        return [{"path": "/ws/notes", "handler": self.notes_socket, "require_auth": True}]

    async def notes_socket(self, websocket, current_user: dict):
        await websocket.accept()
        mine = self.sockets.setdefault(current_user["id"], set())
        mine.add(websocket)
        try:
            while True:
                await websocket.receive_text()
        except WebSocketDisconnect:
            pass
        finally:
            mine.discard(websocket)

    async def on_envelope(self, envelope: dict):
        if envelope.get("kind") != "note_added":
            return
        for socket in list(self.sockets.get(envelope["owner_id"], ())):
            await socket.send_json({"changed": "notes"})

    async def add_note(self, current_user: dict, request: dict = None):
        ...   # the check and the INSERT, as in the second example
        await notification_bus.deliver({"kind": "note_added", "owner_id": current_user["id"]})
        await emit_event("note_added", f"{current_user['username']} added a note",
                         {"owner_id": current_user["id"]})
        return {"ok": True}

Without notification_channel in the settings there is no bus at all, and deliver() hands nothing to anybody. On SQLite the bus exists but does not cross processes — for a single replica that is enough.

05

Sign-in through a company provider

An account is created only for verified addresses of your own domain; everyone else at the same provider is refused. The provider's roles from roles_claim are mapped onto the application's permissions.

import os

from keepup import KeepupSettings
from keepup.settings import OidcSettings
from keepup_auth.oidc_policy import create_if_email_domain

settings = KeepupSettings(
    title="Notes",
    project_name="notes",
    plugins_dir=PLUGINS_DIR,
    oidc=OidcSettings(
        issuer="https://id.example.com/realms/staff",
        client_id="notes",
        client_secret=os.environ["OIDC_CLIENT_SECRET"],
        redirect_uri="https://notes.example.com/api/auth/oidc/callback",
        roles_claim="groups",
        after_login_path="/selfcare",
    ),
    oidc_account_policy=create_if_email_domain(["example.com"]),
)

06

Your own rules: password, audit, origins

What the framework cannot know, the application states itself: which password will do, which request fields the audit may keep in full, and which origins may read the answers.

from keepup_audit.audit import names_only

MIN_PASSWORD_LENGTH = 12
#: Fields that are not secret and help when reading an incident.
AUDIT_PLAIN_FIELDS = {"limit", "note_id"}


def password_rule(password: str, username: str):
    if len(password) < MIN_PASSWORD_LENGTH:
        return f"The password is shorter than {MIN_PASSWORD_LENGTH} characters"
    if username and username.lower() in password.lower():
        return "The password contains the user name"
    return None


def audit_redaction(body):
    if isinstance(body, dict):
        return {key: value if key in AUDIT_PLAIN_FIELDS else names_only(value)
                for key, value in body.items()}
    return names_only(body)


settings = KeepupSettings(
    title="Notes",
    project_name="notes",
    plugins_dir=PLUGINS_DIR,
    password_rule=password_rule,
    audit_redaction=audit_redaction,
    cors_origins=("https://notes.example.com",),
)