Todos¶
A multi-user todo list: persistence, sign-in, a service that is also a REST API, and strict per-user isolation.
todos/
├── app.py # composition root
├── settings.py # DATABASE_URL
├── models/ # database.py, todo.py (Todo, TodoOut, NewTodo), account.py (SignIn)
├── services/ # todos.py (TodoService)
└── views/
├── components/ # todo_item.py (TodoItem)
├── pages/ # todos.py (Todos), login.py (Login)
└── styles/
1. Settings and the database¶
settings.py¶
"""Settings, read once from the environment."""
import os
from pathlib import Path
PROJECT_DIR = Path(__file__).resolve().parent
DATABASE_URL = os.environ.get("DATABASE_URL", f"sqlite+aiosqlite:///{(PROJECT_DIR / 'todos.db').as_posix()}")
models/database.py¶
"""One engine for the app, and the declarative base for its tables."""
from sqlalchemy.orm import DeclarativeBase
from rational.backend import Database
import settings
class Base(DeclarativeBase):
"""Any declarative base works; ``rational.backend.Model`` is a ready-made one."""
database = Database(settings.DATABASE_URL)
async def create_tables() -> None:
"""Development convenience: use Alembic migrations in production."""
await database.create_all(Base)
2. The service owns the rules¶
services/todos.py¶
"""Every database operation lives here, scoped to the signed-in user.
The ``@remote`` methods are also a JSON API under ``/api/todo/...``.
"""
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession
from rational import NotFound, RemoteError
from rational.backend import Service, remote
from models import Todo, TodoOut
class TodoService(Service):
"""The signed-in user's todos."""
@remote(method="GET", auth="required")
async def list_todos(self) -> list[TodoOut]:
"""List your todos, oldest first."""
async with self.db.session() as session:
rows = await session.scalars(select(Todo).where(Todo.owner == self.user).order_by(Todo.id))
return [TodoOut.model_validate(t) for t in rows]
@remote(auth="required")
async def add(self, title: str) -> TodoOut:
"""Create a todo."""
title = title.strip()
if not title:
raise RemoteError("A todo needs a title", code="invalid")
async with self.db.transaction() as session:
todo = Todo(owner=self.user, title=title[:200])
session.add(todo)
await session.flush()
return TodoOut.model_validate(todo)
@remote(auth="required")
async def set_done(self, todo_id: int, done: bool) -> TodoOut:
"""Mark a todo as done or not done."""
async with self.db.transaction() as session:
todo = await self._owned(session, todo_id)
todo.done = done
return TodoOut.model_validate(todo)
@remote(auth="required")
async def remove(self, todo_id: int) -> None:
"""Delete a todo."""
async with self.db.transaction() as session:
await session.delete(await self._owned(session, todo_id))
@remote(auth="required")
async def clear_done(self) -> int:
"""Delete every completed todo; returns how many were removed."""
async with self.db.transaction() as session:
result = await session.execute(delete(Todo).where(Todo.owner == self.user, Todo.done.is_(True)))
return int(result.rowcount or 0)
async def _owned(self, session: AsyncSession, todo_id: int) -> Todo:
todo = await session.get(Todo, todo_id)
if todo is None or todo.owner != self.user: # never reveal that other users' rows exist
raise NotFound("No such todo")
return todo
Every query filters by self.user. _owned answers “not found” for other users’ todos,
so the API never even confirms they exist. Each @remote method becomes a typed endpoint:
GET /api/todo/list-todos, POST /api/todo/add and so on, all in Swagger UI at /api/docs.
3. A component with a callback prop¶
class TodoItem(Component):
"""One row. ``changed`` is a callback prop: the list reloads after a change."""
todo: TodoOut
changed: Callable[[], Awaitable[None]]
async def toggle(self, done: bool):
await self.use(TodoService).set_done(self.todo.id, done)
await self.changed()
async def remove(self):
await self.use(TodoService).remove(self.todo.id)
await self.changed()
def render(self):
return h(
"li",
Checkbox(
self.todo.title,
checked=self.todo.done,
on_change=self.toggle,
style=Style(grow=1) | (DONE if self.todo.done else None),
),
Button(
"Delete",
variant="ghost",
size="sm",
aria_label=f"Delete {self.todo.title}",
on_click=[js.confirm(f"Delete “{self.todo.title}”?"), self.remove], # asks in the browser first
),
style=TODO_ROW,
)
changed is a prop holding the page’s reload method: the item tells its parent that
something changed. on_click=[js.confirm(...), self.remove] asks in the browser first; if the
user cancels, no request is sent.
4. The page¶
@router.page("/", auth="required")
class Todos(Page):
title = "Todos · Rational Web"
todos = State(factory=list)
filter = State("all")
form_version = State(0)
async def mount(self):
await self.reload()
async def reload(self):
self.todos = await self.use(TodoService).list_todos()
async def add(self, form: NewTodo):
await self.use(TodoService).add(form.title)
self.form_version += 1 # a form with a new key is a fresh form: the input is cleared
await self.reload()
self.run_js(js.focus("#new-todo [name=title]"))
async def clear_done(self):
await self.use(TodoService).clear_done()
await self.reload()
def sign_out(self):
self.context.session.logout()
self.navigate("/login")
def show(self, name: str):
return lambda: setattr(self, "filter", name)
def render(self):
visible = [t for t in self.todos if FILTERS[self.filter](t)]
remaining = sum(not t.done for t in self.todos)
filters = Row(
*[
Button(
name.capitalize(),
size="sm",
variant="secondary" if self.filter == name else "ghost",
aria_pressed="true" if self.filter == name else "false",
on_click=self.show(name),
)
for name in FILTERS
],
gap="xs",
role="group",
aria_label="Filter",
)
return Container(
Header(
Heading("Todos", level=1),
Spacer(),
Text(f"Signed in as {self.context.user}", muted=True, size="sm"),
Button("Sign out", variant="outline", size="sm", on_click=self.sign_out),
style=TOOLBAR,
),
Card(
Form(
Input("New todo", name="title", placeholder="What needs doing?", hide_label=True, required=True),
Button("Add", type="submit"),
on_submit=self.add,
id="new-todo",
key=f"form-{self.form_version}",
style=NEW_TODO_FORM,
),
filters,
h("ul", *[TodoItem(todo=t, changed=self.reload, key=t.id) for t in visible], style=TODO_LIST),
Text("Nothing here yet.", muted=True) if not visible else None,
Row(
Text(f"{remaining} item{'s' if remaining != 1 else ''} left", muted=True, size="sm"),
Spacer(),
Button(
"Clear completed",
variant="ghost",
size="sm",
on_click=self.clear_done,
disabled=remaining == len(self.todos),
),
),
),
tag="main",
max_width=640,
style=PAGE,
)
auth="required"sends anonymous visitors to/login?next=/.key=f"form-{self.form_version}"clears the form after adding.self.run_js(js.focus(...))puts the cursor back in the input once the page has updated.
5. Composition¶
app.py¶
"""Todos: the composition root.
Run: pip install "rational[db]"
python app.py --dev # hot reload, readable errors
python app.py --prod # production server
"""
from rational import App
from models import create_tables, database
from services import TodoService
from views.pages import router
from views.styles import THEME
app = App(title="Todos", theme=THEME, database=database, login_path="/login")
app.include(router) # the pages
app.service(TodoService) # also expose the service as a JSON API: /api/todo/... (docs at /api/docs)
app.on_startup(create_tables)
if __name__ == "__main__":
app.run()