Одно приложение, шаг за шагом

01

Приложение с плагинами

Каталог плагинов вычисляется от файла, а не от рабочего каталога. Таблицы, объявленные плагинами, ядро создаёт само при create_app (get_declared_tables()); первые учётные записи создаёт keepup-auth. Канал шины понадобится в четвёртом примере.

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

Плагин со своей таблицей и маршрутами

Таблица объявлена один раз и создаётся при каждом старте, если её нет. Маршруты — словари; маска params проверяет и приводит параметры до вызова обработчика. Запросы к базе синхронные, поэтому уходят в поток и не держат цикл событий.

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)   # свои таблицы плагин ещё и объявляет: get_declared_tables()
        return True

    def get_handlers(self) -> Dict[str, Any]:
        return {"count_notes": self.count_notes}   # для других плагинов

    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"]

Проверка из командной строки — после входа в панель или с токеном:

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: вне маски

03

Фоновая работа в одной реплике

Цикл запускается в каждой реплике, но чистка идёт только в той, что взяла блокировку; остальные пропускают проход. Блокировка продлевается, пока работа идёт, так что длинный проход не перехватят.

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)   # свои таблицы плагин ещё и объявляет: 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, если чистит другая реплика

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

04

События, шина и живой сокет

Сокет пользователя держит одна реплика, а заметку могли добавить через другую. deliver() отдаёт конверт всем репликам, включая свою; каждая пишет в сокеты, которые держит сама. Конверт называет заметку, а не несёт её: доставка по возможности, и клиент перечитывает список сам.

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 -> сокеты В ЭТОЙ реплике

    async def initialize(self) -> bool:
        tables.ensure_tables(notes)   # свои таблицы плагин ещё и объявляет: 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]:
        # У сокетов вход по умолчанию выключен -- включаем явно.
        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):
        ...   # проверка и INSERT, как во втором примере
        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}

Без notification_channel в настройках шины нет вовсе, и deliver() никому ничего не раздаёт. На SQLite шина есть, но между процессами не ходит — для одной реплики этого достаточно.

05

Вход через корпоративный провайдер

Учётная запись создаётся только для подтверждённых адресов своего домена; всем остальным у того же провайдера — отказ. Роли провайдера из roles_claim сопоставляются с правами приложения.

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

Свои правила: пароль, аудит, источники

То, что каркас не может знать, приложение говорит само: какой пароль годится, какие поля запроса можно хранить в аудите целиком и каким источникам можно читать ответы.

from keepup_audit.audit import names_only

MIN_PASSWORD_LENGTH = 12
#: Поля, которые не секретны и нужны при разборе инцидента.
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",),
)