Scaffold project, data model and artwork

Establish the LLeMbas foundation: FastAPI/Jinja/SQLite layout, the ORM
schema, and the original SVG identity.

Notable decisions, all recorded in comments at the point they matter:

- No Alembic. SQLite only, schema created at startup, so models carry a
  few columns nothing reads yet (Message.parent_id for branching,
  content_parts_json for multimodal turns). Adding them later to a live
  database without migrations is the painful path.
- Sessions are server-side rows keyed by a SHA-256 of the cookie value,
  not JWTs, so logout and bans revoke access immediately.
- Upstream API keys are Fernet-encrypted with a key derived from
  LEMBAS_SECRET_KEY. decrypt() fails soft to "" so rotating the secret
  degrades to re-entering keys rather than crashing the admin UI.
- Artwork is generated by scripts/build_artwork.py rather than hand-drawn
  per file: the mallorn leaf appears in the icon, favicon, lockup and
  banner, and one source is the only way those stay in sync. The wordmark
  is Source Serif 4 (OFL) converted to outlines, because a README banner
  cannot load a webfont and <text> would render in whatever serif the
  viewer happens to have.
- Icons live in a template partial, not assets/, because same-document
  <use href="#id"> is universally supported and the cross-document form
  is not.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaroslav Beneš
2026-07-21 10:34:48 +02:00
parent 0665027bc6
commit 5ef2af6a9f
32 changed files with 2041 additions and 0 deletions
View File
+54
View File
@@ -0,0 +1,54 @@
"""Declarative base and column conventions shared by every model.
There is no Alembic in this project (SQLite only, schema created at startup).
That makes adding a column to an existing deployment a manual chore, so models
carry a few forward-looking columns that are not read yet -- see the notes on
``Message.parent_id`` and ``Message.content_parts_json``.
"""
from __future__ import annotations
import uuid
from datetime import UTC, datetime
from sqlalchemy import DateTime, MetaData, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
# Explicit naming convention so constraints have stable, predictable names.
NAMING_CONVENTION = {
"ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s",
}
def new_id() -> str:
"""Primary keys are UUID4 hex strings: URL-safe and non-enumerable."""
return uuid.uuid4().hex
def utcnow() -> datetime:
return datetime.now(UTC)
class Base(DeclarativeBase):
metadata = MetaData(naming_convention=NAMING_CONVENTION)
class UUIDPrimaryKey:
"""Mixin: opaque string primary key generated in Python."""
id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id)
class Timestamps:
"""Mixin: creation and modification times, both timezone-aware UTC."""
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=utcnow, nullable=False
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=utcnow, onupdate=utcnow, nullable=False
)
+45
View File
@@ -0,0 +1,45 @@
"""All ORM models.
Importing this package registers every table on ``Base.metadata``, which is
what ``init_db()`` relies on to create the schema at startup. Any new model
module must be imported here or its table will silently never be created.
"""
from lembas.db.models.chat import (
ROLE_ASSISTANT,
ROLE_SYSTEM,
ROLE_TOOL,
ROLE_USER,
Chat,
Folder,
Message,
)
from lembas.db.models.connection import Connection, Model
from lembas.db.models.setting import Setting
from lembas.db.models.user import (
ROLE_ADMIN,
ROLE_PENDING,
Group,
Session,
User,
user_groups,
)
__all__ = [
"ROLE_ADMIN",
"ROLE_ASSISTANT",
"ROLE_PENDING",
"ROLE_SYSTEM",
"ROLE_TOOL",
"ROLE_USER",
"Chat",
"Connection",
"Folder",
"Group",
"Message",
"Model",
"Session",
"Setting",
"User",
"user_groups",
]
+117
View File
@@ -0,0 +1,117 @@
"""Folders, chats and messages."""
from __future__ import annotations
from typing import Any
from sqlalchemy import Boolean, ForeignKey, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from lembas.db.base import Base, Timestamps, UUIDPrimaryKey
from lembas.db.types import JSONDict, JSONList
ROLE_SYSTEM = "system"
ROLE_USER = "user"
ROLE_ASSISTANT = "assistant"
ROLE_TOOL = "tool"
class Folder(UUIDPrimaryKey, Timestamps, Base):
"""A user-owned, arbitrarily nested container for chats."""
__tablename__ = "folders"
user_id: Mapped[str] = mapped_column(
String(32), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
parent_id: Mapped[str | None] = mapped_column(
String(32), ForeignKey("folders.id", ondelete="CASCADE")
)
name: Mapped[str] = mapped_column(String(200), nullable=False)
position: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
collapsed: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
children: Mapped[list[Folder]] = relationship(
back_populates="parent",
cascade="all, delete-orphan",
order_by="Folder.position, Folder.name",
)
parent: Mapped[Folder | None] = relationship(back_populates="children", remote_side="Folder.id")
chats: Mapped[list[Chat]] = relationship(back_populates="folder")
def __repr__(self) -> str:
return f"<Folder {self.name}>"
class Chat(UUIDPrimaryKey, Timestamps, Base):
__tablename__ = "chats"
user_id: Mapped[str] = mapped_column(
String(32), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
# Deleting a folder keeps its chats; they fall back to the unfiled list.
folder_id: Mapped[str | None] = mapped_column(
String(32), ForeignKey("folders.id", ondelete="SET NULL"), index=True
)
title: Mapped[str] = mapped_column(String(300), default="New chat")
# Set once the model writes the first reply, so auto-titling only runs once.
title_generated: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
# Denormalised rather than a foreign key: chat history must survive an admin
# deleting a connection or a model disappearing upstream.
model_id: Mapped[str] = mapped_column(String(300), default="")
connection_id: Mapped[str | None] = mapped_column(
String(32), ForeignKey("connections.id", ondelete="SET NULL")
)
system_prompt: Mapped[str] = mapped_column(Text, default="")
params_json: Mapped[dict[str, Any]] = mapped_column(JSONDict, default=dict)
pinned: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
archived: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
folder: Mapped[Folder | None] = relationship(back_populates="chats")
messages: Mapped[list[Message]] = relationship(
back_populates="chat",
cascade="all, delete-orphan",
order_by="Message.created_at",
)
def __repr__(self) -> str:
return f"<Chat {self.title!r}>"
class Message(UUIDPrimaryKey, Timestamps, Base):
__tablename__ = "messages"
chat_id: Mapped[str] = mapped_column(
String(32), ForeignKey("chats.id", ondelete="CASCADE"), nullable=False, index=True
)
# Reserved for conversation branching (edit a message, regenerate a reply
# and keep both). Nothing reads it yet; it exists now because retrofitting a
# column onto a live SQLite database without migrations is painful.
parent_id: Mapped[str | None] = mapped_column(String(32), ForeignKey("messages.id"))
role: Mapped[str] = mapped_column(String(16), nullable=False)
content: Mapped[str] = mapped_column(Text, default="")
# Reserved for multimodal turns: [{"type": "image_url", ...}, ...].
# Plain-text messages leave this empty and use `content`.
content_parts_json: Mapped[list[Any]] = mapped_column(JSONList, default=list)
model_id: Mapped[str] = mapped_column(String(300), default="")
tool_calls_json: Mapped[list[Any]] = mapped_column(JSONList, default=list)
usage_json: Mapped[dict[str, Any]] = mapped_column(JSONDict, default=dict)
# Non-empty when generation failed. Rendered as a styled error in the
# thread so a failed turn is never an unexplained blank bubble.
error: Mapped[str] = mapped_column(Text, default="")
# False while a reply is still streaming; flipped when the stream ends.
complete: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
chat: Mapped[Chat] = relationship(back_populates="messages")
def __repr__(self) -> str:
return f"<Message {self.role} {self.content[:40]!r}>"
+82
View File
@@ -0,0 +1,82 @@
"""OpenAI-compatible endpoint connections and their discovered models."""
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship
from lembas.db.base import Base, Timestamps, UUIDPrimaryKey
from lembas.db.types import JSONDict
class Connection(UUIDPrimaryKey, Timestamps, Base):
"""A configured upstream endpoint speaking the OpenAI HTTP API.
Works for api.openai.com as well as LM Studio, vLLM, llama.cpp, Ollama's
compatibility layer, OpenRouter, and anything else exposing /v1.
"""
__tablename__ = "connections"
name: Mapped[str] = mapped_column(String(120), nullable=False)
base_url: Mapped[str] = mapped_column(String(500), nullable=False)
# Fernet ciphertext, never the raw key. See lembas.services.crypto.
# Empty string is legitimate: local endpoints often need no auth at all.
api_key_encrypted: Mapped[str] = mapped_column(Text, default="")
enabled: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
position: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
# Extra headers merged into every request (e.g. OpenRouter's HTTP-Referer).
extra_headers_json: Mapped[dict[str, Any]] = mapped_column(JSONDict, default=dict)
# Result of the most recent "Test & refresh", surfaced in the admin list.
last_checked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
last_error: Mapped[str] = mapped_column(Text, default="")
models: Mapped[list[Model]] = relationship(
back_populates="connection",
cascade="all, delete-orphan",
order_by="Model.model_id",
)
def __repr__(self) -> str:
return f"<Connection {self.name} {self.base_url}>"
class Model(UUIDPrimaryKey, Timestamps, Base):
"""A model advertised by a connection, cached locally.
Cached rather than fetched live so the chat UI stays responsive and keeps
working when an endpoint is briefly unreachable. Refreshed on demand from
the admin screen.
"""
__tablename__ = "models"
__table_args__ = (UniqueConstraint("connection_id", "model_id"),)
connection_id: Mapped[str] = mapped_column(
String(32), ForeignKey("connections.id", ondelete="CASCADE"), nullable=False, index=True
)
model_id: Mapped[str] = mapped_column(String(300), nullable=False)
display_name: Mapped[str] = mapped_column(String(300), default="")
enabled: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
# Endpoints do not reliably advertise capabilities, so these are admin
# overrides consumed by later passes (vision uploads, tool calling).
capabilities_json: Mapped[dict[str, Any]] = mapped_column(JSONDict, default=dict)
# Default sampling params applied to new chats using this model.
params_json: Mapped[dict[str, Any]] = mapped_column(JSONDict, default=dict)
connection: Mapped[Connection] = relationship(back_populates="models")
@property
def label(self) -> str:
return self.display_name or self.model_id
def __repr__(self) -> str:
return f"<Model {self.model_id}>"
+28
View File
@@ -0,0 +1,28 @@
"""Instance-wide settings, stored as a key/value table."""
from __future__ import annotations
from typing import Any
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from lembas.db.base import Base, Timestamps
from lembas.db.types import JSONDict
class Setting(Timestamps, Base):
"""One row per settings group, value is an arbitrary JSON object.
A key/value table rather than a wide typed table: admin settings grow with
every feature (tools, agents, image generation) and adding a column to a
live SQLite database without migrations is exactly what this avoids.
"""
__tablename__ = "settings"
key: Mapped[str] = mapped_column(String(120), primary_key=True)
value: Mapped[dict[str, Any]] = mapped_column(JSONDict, default=dict)
def __repr__(self) -> str:
return f"<Setting {self.key}>"
+89
View File
@@ -0,0 +1,89 @@
"""Users, groups and login sessions."""
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Index, String, Table, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from lembas.db.base import Base, Timestamps, UUIDPrimaryKey
from lembas.db.types import JSONDict
# Roles are a simple ordered ladder rather than a permission matrix. Groups
# (below) carry finer-grained permissions once the users/groups UI lands.
ROLE_ADMIN = "admin"
ROLE_USER = "user"
ROLE_PENDING = "pending" # registered but awaiting admin approval
user_groups = Table(
"user_groups",
Base.metadata,
Column("user_id", String(32), ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
Column("group_id", String(32), ForeignKey("groups.id", ondelete="CASCADE"), primary_key=True),
)
class User(UUIDPrimaryKey, Timestamps, Base):
__tablename__ = "users"
email: Mapped[str] = mapped_column(String(320), unique=True, nullable=False, index=True)
name: Mapped[str] = mapped_column(String(120), nullable=False)
password_hash: Mapped[str] = mapped_column(Text, nullable=False)
role: Mapped[str] = mapped_column(String(16), default=ROLE_USER, nullable=False)
active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
# Per-user preferences: theme, default model, composer behaviour, etc.
settings_json: Mapped[dict[str, Any]] = mapped_column(JSONDict, default=dict)
groups: Mapped[list[Group]] = relationship(secondary=user_groups, back_populates="users")
sessions: Mapped[list[Session]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
@property
def is_admin(self) -> bool:
return self.role == ROLE_ADMIN
def __repr__(self) -> str:
return f"<User {self.email} role={self.role}>"
class Group(UUIDPrimaryKey, Timestamps, Base):
"""A named set of users. Permissions are enforced once the RBAC pass lands."""
__tablename__ = "groups"
name: Mapped[str] = mapped_column(String(120), unique=True, nullable=False)
description: Mapped[str] = mapped_column(Text, default="")
permissions_json: Mapped[dict[str, Any]] = mapped_column(JSONDict, default=dict)
users: Mapped[list[User]] = relationship(secondary=user_groups, back_populates="groups")
class Session(UUIDPrimaryKey, Timestamps, Base):
"""Server-side login session.
Sessions live in the database rather than in a signed JWT so that logging
out, banning a user, or rotating a device actually revokes access
immediately instead of waiting for a token to expire.
"""
__tablename__ = "sessions"
user_id: Mapped[str] = mapped_column(
String(32), ForeignKey("users.id", ondelete="CASCADE"), nullable=False
)
# SHA-256 of the cookie value. The raw token is shown to the browser once
# and never stored, so a database leak does not hand over live sessions.
token_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
user_agent: Mapped[str] = mapped_column(Text, default="")
ip_address: Mapped[str] = mapped_column(String(45), default="")
user: Mapped[User] = relationship(back_populates="sessions")
Index("ix_sessions_user_id", Session.user_id)
+102
View File
@@ -0,0 +1,102 @@
"""Engine, session factory and startup schema creation."""
from __future__ import annotations
import logging
from collections.abc import Iterator
from contextlib import contextmanager
from sqlalchemy import Engine, create_engine, event
from sqlalchemy.orm import Session, sessionmaker
from lembas.config import settings
from lembas.db.base import Base
log = logging.getLogger(__name__)
_engine: Engine | None = None
_SessionFactory: sessionmaker[Session] | None = None
@event.listens_for(Engine, "connect")
def _configure_sqlite(dbapi_connection, connection_record) -> None: # noqa: ANN001
"""Apply the pragmas SQLite needs to behave under a concurrent web server.
- WAL lets readers proceed while a write is in flight, which matters because
a streaming reply holds a write open for the length of the generation.
- foreign_keys is OFF by default in SQLite, so every ondelete= in the models
would be decoration without this.
- busy_timeout makes concurrent writers wait rather than fail instantly.
"""
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA journal_mode=WAL")
cursor.execute("PRAGMA foreign_keys=ON")
cursor.execute("PRAGMA busy_timeout=5000")
cursor.execute("PRAGMA synchronous=NORMAL")
cursor.close()
def get_engine() -> Engine:
global _engine
if _engine is None:
settings.ensure_dirs()
_engine = create_engine(
f"sqlite:///{settings.db_path}",
# FastAPI runs sync endpoints in a threadpool, so a connection can
# legitimately be used from a thread other than the one that made it.
connect_args={"check_same_thread": False},
echo=False,
future=True,
)
return _engine
def get_session_factory() -> sessionmaker[Session]:
global _SessionFactory
if _SessionFactory is None:
_SessionFactory = sessionmaker(
bind=get_engine(),
autoflush=False,
expire_on_commit=False,
)
return _SessionFactory
def init_db() -> None:
"""Create any missing tables.
This is ``CREATE TABLE IF NOT EXISTS`` only -- it never alters an existing
table. There is no migration tool in this project by design, so changing a
column on a model requires migrating the database by hand.
"""
import lembas.db.models # noqa: F401 (registers tables on the metadata)
Base.metadata.create_all(bind=get_engine())
log.debug("schema ensured at %s", settings.db_path)
@contextmanager
def session_scope() -> Iterator[Session]:
"""Transactional scope for background work and CLI commands.
Request handlers should use the `db` dependency in lembas.api.deps instead.
"""
factory = get_session_factory()
session = factory()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
def reset_engine() -> None:
"""Drop cached engine/factory. Used by tests to rebind to a temp database."""
global _engine, _SessionFactory
if _engine is not None:
_engine.dispose()
_engine = None
_SessionFactory = None
+14
View File
@@ -0,0 +1,14 @@
"""Reusable column types.
SQLite stores JSON as text. Wrapping the JSON type in SQLAlchemy's mutation
tracking means ``obj.settings_json["theme"] = "shire"`` marks the row dirty --
without it, in-place edits of a dict column are silently dropped on flush.
"""
from __future__ import annotations
from sqlalchemy import JSON
from sqlalchemy.ext.mutable import MutableDict, MutableList
JSONDict = MutableDict.as_mutable(JSON)
JSONList = MutableList.as_mutable(JSON)