# Zeython Docs > Zeython — an async-first, batteries-included MVC framework for Python Zeython is an async-first, batteries-included MVC framework for Python: a dependency injection container, an Active-Record-style async ORM, session and API-token auth, RBAC authorization, queues, WebSockets, and a Laravel-style CLI, built on Starlette and SQLAlchemy 2.0. # Getting Started # Zeython Zeython is an async-first, batteries-included MVC framework for Python, built on [Starlette](https://www.starlette.io/) and [SQLAlchemy 2.0](https://docs.sqlalchemy.org/). It gives you a dependency injection container, a service-provider boot lifecycle, an Active-Record-style async ORM, and a Laravel-style `zeython` CLI — the pieces most hand-rolled Python web projects end up building themselves, done once and done well. ## Why Zeython Python has excellent web *libraries* (Starlette, SQLAlchemy, Alembic, uvicorn) but assembling them into a coherent, opinionated application structure is left as an exercise to every team. Zeython is that assembly: a real framework with conventions, not a template you copy and diverge from. - **Async all the way down.** Request handling, the ORM, and migrations are async from the start — no bolted-on `asyncio.run` calls. - **Request-scoped database sessions**, not a global session shared across requests or a fresh session hand-rolled per call. - **Convention over configuration.** Controllers, models, and routes live in predictable places (`app/Controllers`, `app/Models`, `routes/`). - **A real CLI.** `zeython new`, `zeython serve`, `zeython make:*`, and `zeython db:*` cover the whole day-to-day loop. ## Where to start - **Getting Started** ______________________________________________________________________ Install the framework and scaffold your first project in under a minute. [Getting Started](https://zeython.zaber.dev/docs/getting-started/index.md) - **Build TaskFlow** ______________________________________________________________________ A guided, six-part tutorial from an empty scaffold to a tested, authenticated API — the fastest way to actually learn the framework. [Start the tutorial](https://zeython.zaber.dev/docs/tutorial/index.md) - **Architecture** ______________________________________________________________________ How the container, service providers, router, and async ORM fit together. [Read Architecture](https://zeython.zaber.dev/docs/architecture/index.md) - **API Reference** ______________________________________________________________________ Every public class and function, generated straight from docstrings. [Browse the reference](https://zeython.zaber.dev/docs/reference/index.md) Everything else — database, security, background jobs, deployment — is one click away in the navigation above. ## Contributing See [CONTRIBUTING.md](https://github.com/zaber-dev/Zeython/blob/main/CONTRIBUTING.md) in the repository root. # Getting Started ## Requirements - Python 3.11+ ## Install ```bash python -m venv .venv source .venv/bin/activate pip install zeython ``` ## Scaffold a project ```bash zeython new "My Blog" cd my_blog pip install -e . cp .env.example .env ``` This generates: ```text my_blog/ ├── app/ │ ├── Controllers/ # request handlers │ ├── Models/ # async Active Record models │ └── Middleware/ # ASGI middleware ├── routes/ │ └── web.py # route definitions ├── migrations/ # Alembic migrations ├── tests/ ├── main.py # application entry point ├── alembic.ini └── .env.example ``` ## Run it ```bash zeython serve ``` Visit `http://127.0.0.1:8000` — you should see a JSON welcome message. Try registering a user and listing them: ```bash curl -X POST http://127.0.0.1:8000/register \ -H 'Content-Type: application/json' \ -d '{"name": "Ada", "email": "ada@example.com", "password": "hunter2"}' curl http://127.0.0.1:8000/users ``` `/users` is paginated (see [Database & Migrations](https://zeython.zaber.dev/docs/database/#pagination)) — the response is `{"items": [...], "page": 1, "total": 1, ...}`, not a bare array. Getting a database error? A fresh scaffold ships with the `User` model but no migration file yet — generate and apply the initial one first: ```bash zeython db revision -m "create users table" zeython db migrate ``` ## Next steps - Follow the [Tutorial](https://zeython.zaber.dev/docs/tutorial/index.md) to build a real, tested, multi-model app from this same starting point — the fastest way to actually learn the framework. - Read [Architecture](https://zeython.zaber.dev/docs/architecture/index.md) to understand the container, service providers, and router. - Read [Database & Migrations](https://zeython.zaber.dev/docs/database/index.md) to add your own models. - Read [CLI Reference](https://zeython.zaber.dev/docs/cli/index.md) for the full list of `zeython make:*` generators. # Architecture ## The pieces | Component | Module | Role | | --------------------------------------------- | ------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `Application` | `zeython.application` | The ASGI entry point. Owns the container, config, and router; boots service providers; builds the underlying Starlette app lazily on first use. | | `Container` | `zeython.container` | A type-hint-driven dependency injection container: `bind`, `singleton`, `instance`, `make`, and `call` (autowired function invocation). | | `ServiceProvider` | `zeython.providers` | The seam where cross-cutting concerns register bindings (`register()`) and wire themselves up (`boot()`). | | `Router` | `zeython.routing` | Decorator-based routing (`@app.get(...)`), route groups (`include`), and RESTful `resource()` registration — compiles to Starlette `Route`/`Mount` objects. | | `Model` | `zeython.db.Model` | An async Active-Record base class: `create`, `find`, `all`, `find_by`, `paginate` (see [Database & Migrations](https://zeython.zaber.dev/docs/database/#pagination)), `save`, `update`, `delete` (soft by default), `to_dict`, plus declarative validation via `__rules__` (see [Validation](https://zeython.zaber.dev/docs/validation/index.md)), overridable lifecycle hooks (`creating`/`created`/`updating`/`updated`/`deleting`/`deleted`) and `Observer` classes for cross-cutting reactions to them (see [Model Events](https://zeython.zaber.dev/docs/model-events/index.md)), and safe relationship eager-loading via `include=` (see [Relationships](https://zeython.zaber.dev/docs/relationships/index.md)). | | `transaction` | `zeython.db` | A `SAVEPOINT`-scoped nested transaction within the current session, for isolating part of a request's writes without ending it (see [Database & Migrations](https://zeython.zaber.dev/docs/database/#transactions)). | | `Database.read_replica()` | `zeython.db.Database` | Opens a session against a read replica (`DATABASE_READ_URL`) instead of the primary, for a read-heavy path that can tolerate lag -- falls back to the primary if no replica is configured (see [Database & Migrations](https://zeython.zaber.dev/docs/database/#read-replicas)). | | `N1QueryDetectionServiceProvider` | `zeython.n_plus_one` | Warns (dev-only, `APP_DEBUG`) when a request fires the same SQL statement shape suspiciously many times -- the N+1 query pattern (see [Relationships](https://zeython.zaber.dev/docs/relationships/#detecting-n1s-automatically)). | | `Config` | `zeython.config` | Layered `.env` + process-environment configuration with dot-path access (`config.get("database.url")`). | | `Views` | `zeython.views` | Jinja2 rendering by convention from `resources/views/`, bound via `ViewServiceProvider` (see [Views](https://zeython.zaber.dev/docs/views/index.md)). | | `CorsServiceProvider` | `zeython.providers` | Opt-in, `.env`-configured CORS support wrapping Starlette's `CORSMiddleware`. | | `AuthServiceProvider` | `zeython.auth` | Session-based auth: signed-cookie sessions, password hashing, `login`/`logout`/`current_user`/`require_auth`, [CSRF protection](https://zeython.zaber.dev/docs/csrf/index.md) on by default (see [Authentication](https://zeython.zaber.dev/docs/authentication/index.md)). | | `CsrfMiddleware` | `zeython.csrf` | Double-submit-cookie CSRF protection; bundled with `AuthServiceProvider`, or usable standalone (see [CSRF Protection](https://zeython.zaber.dev/docs/csrf/index.md)). | | `SecurityHeadersServiceProvider` | `zeython.security_headers` | Opt-in `X-Frame-Options`/`X-Content-Type-Options`/`Referrer-Policy`/CSP/HSTS response headers, `.env`-configured; not registered by default (see [Security Headers](https://zeython.zaber.dev/docs/security-headers/index.md)). | | `ApiAuthServiceProvider` | `zeython.api_auth` | Stateless bearer-token auth for non-cookie clients: `TokenManager.issue`/`.verify`, `require_api_auth`, signed with `itsdangerous` (no new dependency, no token table) (see [API Authentication](https://zeython.zaber.dev/docs/api-authentication/index.md)). | | `Gate` | `zeython.authorization` | Named authorization abilities: `gate.define(...)`, resource-bound Policy classes (`gate.policy(...)`), a global bypass hook (`gate.before(...)`), role/permission checks (`HasRoles`, `Gate.role(...)`/`Gate.permission(...)`), and `authorize(request, ability, ...)` (403 on a failed check, 401 if not logged in at all) (see [Authorization](https://zeython.zaber.dev/docs/authorization/index.md)). | | `OpenApiServiceProvider` | `zeython.openapi` | Generates an OpenAPI 3.0 document from the app's actually-registered routes and serves it plus a Swagger UI (`/openapi.json`, `/docs`); `@describe(...)` adds a real summary/tags/schema to a route (see [OpenAPI & API Docs](https://zeython.zaber.dev/docs/openapi/index.md)). | | `GzipServiceProvider` / `ETagServiceProvider` | `zeython.gzip` / `zeython.etag` | Response compression and conditional-GET (`ETag`/`If-None-Match` → `304`) support; `API_PROBLEM_JSON=true` switches error responses to RFC 7807's `application/problem+json` shape (see [API Standards](https://zeython.zaber.dev/docs/api-standards/index.md)). | | `PluginServiceProvider` | `zeython.plugins` | Discovers and registers every third-party provider declared under the `zeython.plugins` entry-point group by an installed package -- one line turns on discovery, not one line per plugin (see [Plugins](https://zeython.zaber.dev/docs/plugins/index.md)). | | `LocalizationServiceProvider` | `zeython.localization` | Translation strings loaded from `resources/lang/{locale}.json`; `LocaleMiddleware` resolves each request's locale (`?lang=`, then `Accept-Language`, then a default) and makes it available to `t(request, key, ...)`/a `t` Jinja global without threading a request through every call (see [Localization](https://zeython.zaber.dev/docs/localization/index.md)). | | `AdminServiceProvider` | `zeython.admin` | Generates list/create/edit/delete pages for the models you register, from their own columns -- gated by a required `guard` callable, no default that lets any logged-in user in (see [Admin Panel](https://zeython.zaber.dev/docs/admin/index.md)). | | `TenancyServiceProvider` | `zeython.tenancy` | Row-level multi-tenancy: a model opts in by declaring a `tenant_id` column, and `Model`'s query methods scope to the current request's resolved tenant automatically -- no mixin, no per-query flag (see [Multi-Tenancy](https://zeython.zaber.dev/docs/multi-tenancy/index.md)). | | `Storage` | `zeython.storage` | Backend-agnostic file storage (`LocalStorage` by default, `S3Storage` opt-in) with `store_upload()` for safe, validated uploads (see [File Storage](https://zeython.zaber.dev/docs/storage/index.md)). | | `RateLimiter` | `zeython.rate_limit` | In-memory sliding-window rate limiting by default: `throttle()` per-route guard, opt-in blanket middleware, applied to auth by default; `RedisRateLimiter` for a shared, distributed limit (see [Rate Limiting](https://zeython.zaber.dev/docs/rate-limiting/index.md), [Redis](https://zeython.zaber.dev/docs/redis/index.md)). | | `Cache` | `zeython.cache` | An in-memory TTL cache by default: `get`/`put`/`forget`/`has`/`flush`, plus `remember()` for get-or-compute; `RedisCache` for a cache shared across processes/machines (see [Caching](https://zeython.zaber.dev/docs/caching/index.md), [Redis](https://zeython.zaber.dev/docs/redis/index.md)). | | `Queue` | `zeython.queue` | Background jobs: `Job` + `dispatch()`, `InMemoryQueue` (background task, no lifespan wiring needed) by default, `SyncQueue` for tests, `RedisQueue` (durable, retries with backoff, failed-jobs list, run via `zeython queue work`) for production (see [Background Jobs](https://zeython.zaber.dev/docs/queues/index.md)). | | `Schedule` | `zeython.schedule` | Recurring tasks defined in code: `schedule.call(fn).daily()`/`.cron(...)`, run via `zeython schedule run` (one cron entry, however many tasks) -- not registered by default (see [Scheduling](https://zeython.zaber.dev/docs/scheduling/index.md)). | | `Mailer` | `zeython.mail` | Outbound email: `LogMailer` by default (zero setup), `SmtpMailer` opt-in. A job's `handle()` can declare `mailer: Mailer` and get it autowired (see [Mail](https://zeython.zaber.dev/docs/mail/index.md)). | | `AI` | `zeython.ai` | A swappable LLM client for your own app code: `complete()`, `EchoAI` by default (no credentials), `AnthropicAI` opt-in via the `ai` extra (see [AI](https://zeython.zaber.dev/docs/ai/index.md)). | | MCP server | `zeython.mcp` | Read-only project introspection for AI coding agents (`zeython mcp`): real registered routes, real mapped models, app info, and search over the bundled docs — an opt-in `mcp` extra, not imported by the framework core (see [AI Agents](https://zeython.zaber.dev/docs/ai-agents/index.md)). | | `Command` | `zeython.console` | The `app/Console/Commands/` extension point: one `Command` subclass per file, wired to the app's own container/config, run with `zeython command ` and listed with `zeython commands` (see [Console Commands](https://zeython.zaber.dev/docs/console-commands/index.md)). | | `Factory` | `zeython.database.factory` | Model factories for tests and seeding: `make()`/`create()`/`create_many()`, sequence-based uniqueness, no bundled fake-data dependency (see [Factories & Seeders](https://zeython.zaber.dev/docs/database-seeding/index.md)). | | `Seeder` | `zeython.database.seeder` | The `database/seeders/` extension point: `run()` inserts seed data (typically via a `Factory`), `self.call(...)` composes seeders, run with `zeython db seed` (see [Factories & Seeders](https://zeython.zaber.dev/docs/database-seeding/index.md)). | | `HealthCheckServiceProvider` | `zeython.health` | Registers `/up`: `200`/`503` with a real database connectivity check when `Database` is bound -- what a load balancer or Kubernetes probe expects (see [Health Check](https://zeython.zaber.dev/docs/health-check/index.md)). | | `RequestIdServiceProvider` | `zeython.request_id` | Stamps every request/response with a correlation ID (`X-Request-ID`, honoring one the caller already sent) and threads it into the logging context via `request_id()` and `%(request_id)s`; registered by default (see [Observability](https://zeython.zaber.dev/docs/observability/index.md)). | | `JsonFormatter` | `zeython.logging` | One JSON object per log line instead of the default text line -- `LOG_FORMAT=json` (see [Observability](https://zeython.zaber.dev/docs/observability/index.md)). | | `ErrorMonitoringServiceProvider` | `zeython.error_monitoring` | Reports unhandled request exceptions, exhausted job retries, and raising scheduled tasks to Sentry -- opt-in, requires the `sentry` extra and `SENTRY_DSN` (see [Error Monitoring](https://zeython.zaber.dev/docs/error-monitoring/index.md)). | | `WebSocketHub` | `zeython.websockets` | Real-time handlers via `@app.websocket(...)`, built on Starlette's ASGI-native WebSocket support; `WebSocketHub` tracks connections and broadcasts to them, process-local by default (see [WebSockets](https://zeython.zaber.dev/docs/websockets/index.md)). | ## Logging `Application()` calls `logging.basicConfig()` for you — `APP_DEBUG=true` → root level `DEBUG`, otherwise `INFO` — and quiets a handful of noisy third-party loggers (`aiosqlite`, `sqlalchemy.engine`, `asyncio`) to `WARNING` so they don't drown out your own app's logs. This is skipped entirely if the root logger already has a handler when `Application()` runs (you called `logging.basicConfig()` yourself, or your deployment platform did) — it never overrides an existing setup. Without this, INFO-level logs (including background job failures — see [Background Jobs](https://zeython.zaber.dev/docs/queues/index.md)) are silently dropped, since uvicorn only configures its own logger namespaces. ## Boot lifecycle ```python from zeython import Application, DatabaseServiceProvider, RouteServiceProvider app = Application() app.register(DatabaseServiceProvider) app.register(RouteServiceProvider(app, modules=("routes.web",))) ``` 1. `Application()` loads `Config` from `.env`, and creates a `Container` and `Router`. 1. `app.register(provider)` calls the provider's `register()` immediately — this is where bindings go into the container. `RouteServiceProvider` imports your route modules here, which is how `@app.get("/")` decorators in `routes/web.py` end up registered. 1. On the **first** request (or the first `app.asgi` access), `Application` calls `boot()` on every registered provider, in registration order, then builds the underlying Starlette app from the final route list. Splitting `register()` from `boot()` matters: a provider's `boot()` can safely assume every other provider's bindings already exist, regardless of registration order. `DatabaseServiceProvider`, for example, binds the `Database` object in `register()` but only attaches the request-scoped session middleware in `boot()`. ## Request-scoped database sessions Unlike opening a new session per query, Zeython opens exactly one `AsyncSession` per HTTP request (via `DatabaseSessionMiddleware`), stores it in a `contextvars.ContextVar`, and commits/rolls back automatically when the request finishes. `Model` methods pull that session via `current_session()` — you never pass a session object through your call stack: ```python class User(Model): __tablename__ = "users" name: Mapped[str] = mapped_column(String(255)) email: Mapped[str] = mapped_column(String(255), unique=True) async def show(self, request): user = await User.find(int(request.path_params["id"])) ``` Outside of a request (a script, a test, a REPL), open a session explicitly: ```python async with database.session(): user = await User.create(name="Ada", email="ada@example.com") ``` The whole session commits/rolls back around the request; `transaction()` scopes a `SAVEPOINT` inside it, for a chunk of work that should roll back on its own without ending the request — see [Transactions](https://zeython.zaber.dev/docs/database/#transactions). ## Routing and controllers ```python from zeython import Controller class UserController(Controller): async def index(self, request): ... async def show(self, request): ... async def store(self, request): ... app.router.resource("/users", UserController, only=("index", "show", "store")) ``` `resource()` maps `index`/`store`/`show`/`update`/`destroy` to the conventional REST verbs and paths, the same mapping Laravel and Rails use. Function-based routes via `@app.get("/path")` work exactly like Flask/FastAPI for everything else. # Tutorial # Tutorial: Build TaskFlow [Getting Started](https://zeython.zaber.dev/docs/getting-started/index.md) gets you from zero to a running app in five minutes, but five minutes isn't enough to actually learn a framework — you need to build something. This tutorial builds **TaskFlow**, a small multi-project task tracker, from an empty scaffold to a tested, authenticated API, introducing one concept at a time the way [Django's official tutorial](https://docs.djangoproject.com/en/stable/intro/tutorial01/) or [Laravel Bootcamp](https://bootcamp.laravel.com/) do. By the end you'll have written two real models with a relationship between them, a full set of CRUD routes, an authorization rule that only lets a task's owner delete it, and a test suite that actually exercises all of it — and you'll understand *why* each piece works the way it does, not just that it does. ## What you'll build - **Projects** — a name and a owner. - **Tasks** — belong to a project, have a title and a done flag. - Anyone can list and view projects/tasks; only a logged-in user can create a task, and only the task's own creator can delete it. - A test suite covering the CRUD routes and the authorization rule. ## Prerequisites Python 3.11+, and fifteen minutes. No prior Zeython knowledge assumed — if you haven't yet, skim [Getting Started](https://zeython.zaber.dev/docs/getting-started/index.md) first for the one-paragraph mental model of what `zeython new` gives you; this tutorial re-explains everything as it comes up either way. ## Parts 1. [Setup](https://zeython.zaber.dev/docs/tutorial-1-setup/index.md) — scaffold the project, look around, run it, log in with the built-in User model. 1. [Models](https://zeython.zaber.dev/docs/tutorial-2-models/index.md) — define `Project` and `Task`, add validation, run your first migration. 1. [Controllers & Routes](https://zeython.zaber.dev/docs/tutorial-3-controllers/index.md) — a full CRUD API for tasks, tested with real HTTP requests as you go. 1. [Relationships](https://zeython.zaber.dev/docs/tutorial-4-relationships/index.md) — connect `Task` to `Project`, load them together safely, avoid the #1 async ORM mistake. 1. [Authentication & Authorization](https://zeython.zaber.dev/docs/tutorial-5-auth/index.md) — require login to create a task; only its owner can delete it. 1. [Testing](https://zeython.zaber.dev/docs/tutorial-6-testing/index.md) — a real pytest suite for everything you just built, including the authorization rule. Each part leaves you with a working app — run `zeython serve` and try what you just built before moving on. If something doesn't match what this tutorial says, that's worth stopping on: either you hit a real bug (open an issue, or see [SECURITY.md](https://github.com/zaber-dev/Zeython/blob/main/SECURITY.md) if it's security-relevant) or a step got skipped — every command here is meant to be copy-pasteable and correct against the current release. Start with [Part 1: Setup](https://zeython.zaber.dev/docs/tutorial-1-setup/index.md). # Part 1: Setup ## Scaffold the project ```bash python -m venv .venv && source .venv/bin/activate pip install zeython zeython new "TaskFlow" cd task_flow pip install -e ".[dev]" cp .env.example .env ``` `zeython new` generated a real, runnable project — not a single-file demo. (The `[dev]` extra pulls in `pytest`, `pytest-asyncio`, and `httpx` — the project ships with a passing test in `tests/test_home.py` already, and you'll write more of your own starting in [Part 6](https://zeython.zaber.dev/docs/tutorial-6-testing/index.md).) Look at what's there: ```text task_flow/ ├── app/ │ ├── Controllers/ # request handlers │ ├── Models/ # async Active Record models │ ├── Middleware/ # ASGI middleware │ ├── Providers/ # your own service providers │ ├── Jobs/ # background jobs │ └── Console/Commands/ # custom `zeython command ` CLI commands ├── database/ │ ├── factories/ # model factories for tests/seeding │ └── seeders/ # `zeython db seed` ├── routes/ │ └── web.py # route definitions ├── migrations/ # Alembic migrations ├── tests/ ├── main.py # application entry point -- start here ├── alembic.ini └── .env.example ``` `main.py` is worth opening now. It's not hidden framework magic — it's plain Python that constructs an `Application`, registers the providers your app needs (`AuthServiceProvider`, `DatabaseServiceProvider`, `RouteServiceProvider`, ...), and a dozen more commented out that you can turn on later (rate limiting, an admin panel, localization). Everything your generated app does, it does because a line in `main.py` says so — see [Architecture](https://zeython.zaber.dev/docs/architecture/index.md) when you're ready for the full picture of how the container and providers fit together. For now, the short version: a **service provider** is where a piece of functionality (auth, the database, routing) gets wired into the app, and `main.py` is the list of which pieces you're using. ## Run it A fresh scaffold ships with the `User` model already written but no migration file for it yet — `revision` autogenerates one by diffing your models against the (empty) database, `migrate` applies it: ```bash zeython db revision -m "create users table" zeython db migrate zeython serve ``` Visit `http://127.0.0.1:8000` — you'll get a small JSON welcome payload. `zeython serve` auto-reloads on file changes, so leave it running for the rest of this tutorial. ## Try the auth that's already there A generated project ships with a working `User` model and session authentication out of the box (see [Authentication](https://zeython.zaber.dev/docs/authentication/index.md) for the full picture later) — you get to use it for free instead of building login from scratch: ```bash curl -sS -c cookies.txt -X POST http://127.0.0.1:8000/register \ -H 'Content-Type: application/json' \ -d '{"name": "Ada", "email": "ada@example.com", "password": "hunter2222"}' ``` ```json {"name":"Ada","email":"ada@example.com","id":1,"created_at":"...","updated_at":"...","is_deleted":false,"deleted_at":null} ``` `-c cookies.txt` saves the session cookie `/register` sets — you're now logged in as Ada, and every following `curl` in this tutorial that passes `-b cookies.txt` will be authenticated as her. Confirm it: ```bash curl -sS -b cookies.txt http://127.0.0.1:8000/me ``` ```json {"name":"Ada","email":"ada@example.com","id":1,"created_at":"...","updated_at":"...","is_deleted":false,"deleted_at":null} ``` That's the whole account system you'd otherwise hand-roll — password hashing, a signed session cookie, CSRF protection on every unsafe request from here on (you'll deal with that directly in [Part 5](https://zeython.zaber.dev/docs/tutorial-5-auth/index.md), when it's TaskFlow's own routes that need protecting) — already wired up, already tested, already documented. Next: [Part 2 — Models](https://zeython.zaber.dev/docs/tutorial-2-models/index.md), where TaskFlow actually starts. # Part 2: Models TaskFlow needs two things: **projects** to group work, and **tasks** inside them. Generate both: ```bash zeython make model Project zeython make model Task ``` This created `app/Models/project.py` and `app/Models/task.py`, each with a single placeholder `name` column, and registered them in `app/Models/__init__.py` automatically — a model has to actually be imported somewhere for SQLAlchemy to know it exists, and this saves you remembering to do that by hand every time. Open `app/Models/project.py`: ```python from sqlalchemy import String from sqlalchemy.orm import Mapped, mapped_column from zeython import Model class Project(Model): __tablename__ = "projects" name: Mapped[str] = mapped_column(String(255)) ``` That's a real Active Record model, not a wrapper around one — `Model` is a genuine SQLAlchemy declarative base (see [Database & Migrations](https://zeython.zaber.dev/docs/database/index.md) for the full API), so anything you know from plain SQLAlchemy still applies. What `Model` adds on top: `id`, `created_at`, `updated_at`, and soft-delete (`is_deleted`/`deleted_at`) columns on every model automatically, plus an async Active Record API — `Project.create(...)`, `Project.find(id)`, `Project.all()`, `instance.update(...)`, `instance.delete()`. ## Add validation A project needs a name — enforce it with a declarative rule instead of an `if` statement in every place that creates one: ```python from sqlalchemy import String from sqlalchemy.orm import Mapped, mapped_column from zeython import Model, max_length, required class Project(Model): __tablename__ = "projects" __rules__ = { "name": [required(), max_length(255)], } name: Mapped[str] = mapped_column(String(255)) ``` `__rules__` runs automatically on every `save()` — both `create()` and `update()` — and raises a `ValidationException` (a 422 with a field-by-field error dict) on failure. See [Validation](https://zeython.zaber.dev/docs/validation/index.md) for the full list of built-in rules and how to write your own. Now `app/Models/task.py` — a task has a title and a done flag: ```python from sqlalchemy import Boolean, String from sqlalchemy.orm import Mapped, mapped_column from zeython import Model, max_length, required class Task(Model): __tablename__ = "tasks" __rules__ = { "title": [required(), max_length(255)], } title: Mapped[str] = mapped_column(String(255)) done: Mapped[bool] = mapped_column(Boolean, default=False) ``` (Task doesn't belong to a Project yet — that's the whole subject of [Part 4](https://zeython.zaber.dev/docs/tutorial-4-relationships/index.md). For now they're two independent models, which is deliberately simpler for learning the CRUD basics first.) ## Migrate Zeython's migrations are Alembic under the hood, with autogeneration already wired to your models — you write the model, Alembic diffs it against the database and writes the migration for you: ```bash zeython db revision -m "add projects and tasks" zeython db migrate ``` The first command writes a new file under `migrations/versions/` (open it — it's plain, readable Alembic code, not a black box); the second actually applies it. Confirm the tables exist by creating one directly: ```bash curl -sS -X POST http://127.0.0.1:8000/projects \ -H 'Content-Type: application/json' \ -d '{"name": "Website Redesign"}' ``` That'll 404 — there's no `/projects` route yet. Models don't expose routes on their own; that's next. Continue to [Part 3 — Controllers & Routes](https://zeython.zaber.dev/docs/tutorial-3-controllers/index.md). # Part 3: Controllers & Routes Generate a controller for each model: ```bash zeython make controller Project zeython make controller Task ``` This created `app/Controllers/project_controller.py` and `app/Controllers/task_controller.py`, each a `Controller` subclass with a placeholder `index` method. Replace `app/Controllers/project_controller.py`: ```python from starlette.responses import JSONResponse, Response from zeython import Controller, NotFoundException from app.Models.project import Project class ProjectController(Controller): async def index(self, request): projects = await Project.all() return JSONResponse([project.to_dict() for project in projects]) async def show(self, request): project = await Project.find(int(request.path_params["id"])) if project is None: raise NotFoundException("Project not found") return JSONResponse(project.to_dict()) async def store(self, request): data = await request.json() project = await Project.create(name=data.get("name")) return JSONResponse(project.to_dict(), status_code=201) async def update(self, request): project = await Project.find(int(request.path_params["id"])) if project is None: raise NotFoundException("Project not found") data = await request.json() await project.update(**data) return JSONResponse(project.to_dict()) async def destroy(self, request): project = await Project.find(int(request.path_params["id"])) if project is None: raise NotFoundException("Project not found") await project.delete() return Response(status_code=204) ``` Nothing here is Zeython-specific magic — it's plain async Python calling the Active Record API from [Part 2](https://zeython.zaber.dev/docs/tutorial-2-models/index.md). `NotFoundException` (and its siblings — `ValidationException`, `UnauthorizedException`, `ForbiddenException`) are the framework's way of turning "this went wrong" into the right HTTP status and JSON shape without you writing that translation by hand every time; raise one, the framework handles the response. Do the same for `app/Controllers/task_controller.py`, swapping `Project` for `Task` and `name` for `title` (and pass `done` through on create if you want to set it explicitly — it defaults to `False` from the model either way): ```python from starlette.responses import JSONResponse, Response from zeython import Controller, NotFoundException from app.Models.task import Task class TaskController(Controller): async def index(self, request): tasks = await Task.all() return JSONResponse([task.to_dict() for task in tasks]) async def show(self, request): task = await Task.find(int(request.path_params["id"])) if task is None: raise NotFoundException("Task not found") return JSONResponse(task.to_dict()) async def store(self, request): data = await request.json() task = await Task.create(title=data.get("title")) return JSONResponse(task.to_dict(), status_code=201) async def update(self, request): task = await Task.find(int(request.path_params["id"])) if task is None: raise NotFoundException("Task not found") data = await request.json() await task.update(**data) return JSONResponse(task.to_dict()) async def destroy(self, request): task = await Task.find(int(request.path_params["id"])) if task is None: raise NotFoundException("Task not found") await task.delete() return Response(status_code=204) ``` ## Wire up the routes Open `routes/web.py`, import both controllers, and register a full REST resource for each: ```python from app.Controllers.project_controller import ProjectController from app.Controllers.task_controller import TaskController app.router.resource("/projects", ProjectController) app.router.resource("/tasks", TaskController) ``` `resource()` maps one controller onto the standard five CRUD routes in one line: | Method | Path | Controller method | | ------------- | ---------------- | ----------------- | | `GET` | `/projects` | `index` | | `POST` | `/projects` | `store` | | `GET` | `/projects/{id}` | `show` | | `PUT`/`PATCH` | `/projects/{id}` | `update` | | `DELETE` | `/projects/{id}` | `destroy` | Pass `only=("index", "show")` if you only want a subset — the generated `UserController`/`PostController` in `routes/web.py` already do this for routes that shouldn't exist yet (see the finished file). ## Try it `zeython serve` picked up the changes automatically. Create a project, then a task: ```bash curl -sS -X POST http://127.0.0.1:8000/projects \ -H 'Content-Type: application/json' \ -d '{"name": "Website Redesign"}' ``` ```json {"name":"Website Redesign","id":1,"created_at":"...","updated_at":"...","is_deleted":false,"deleted_at":null} ``` ```bash curl -sS -X POST http://127.0.0.1:8000/tasks \ -H 'Content-Type: application/json' \ -d '{"title": "Design the new homepage"}' curl -sS http://127.0.0.1:8000/tasks curl -sS http://127.0.0.1:8000/tasks/1 curl -sS -X PUT http://127.0.0.1:8000/tasks/1 \ -H 'Content-Type: application/json' \ -d '{"done": true}' curl -sS -X DELETE http://127.0.0.1:8000/tasks/1 -o /dev/null -w '%{http_code}\n' ``` The last command prints `204` — a successful delete with no body. You now have full CRUD for two models. What's missing: a task has no idea which project it belongs to. Continue to [Part 4 — Relationships](https://zeython.zaber.dev/docs/tutorial-4-relationships/index.md). # Part 4: Relationships A task with no project isn't very useful for a *multi-project* tracker. Add the connection to `app/Models/task.py`: ```python from sqlalchemy import Boolean, ForeignKey, String from sqlalchemy.orm import Mapped, mapped_column, relationship from zeython import Model, max_length, required from app.Models.project import Project class Task(Model): __tablename__ = "tasks" __rules__ = { "title": [required(), max_length(255)], } title: Mapped[str] = mapped_column(String(255)) done: Mapped[bool] = mapped_column(Boolean, default=False) project_id: Mapped[int] = mapped_column(ForeignKey("projects.id")) project: Mapped[Project] = relationship(back_populates="tasks") ``` And the other side, in `app/Models/project.py`: ```python from sqlalchemy.orm import Mapped, mapped_column, relationship # ... existing imports ... class Project(Model): # ... existing __tablename__/__rules__/name ... tasks: Mapped[list["Task"]] = relationship(back_populates="project") ``` (The string `"Task"` in the type hint — not the class itself — sidesteps a circular import: `task.py` imports `Project`, so `project.py` can't also import `Task` at module scope. SQLAlchemy resolves the string by class name once both models are loaded.) ## Migrate ```bash zeython db revision -m "add task.project_id" ``` If you followed [Part 3](https://zeython.zaber.dev/docs/tutorial-3-controllers/index.md)'s walkthrough exactly, the task you created and then deleted there is still in the `tasks` table — soft-delete marks it `is_deleted`, it doesn't remove the row — and it has no `project_id`. Open the migration file Alembic just generated under `migrations/versions/`; the new column comes out `nullable=False` with no value for that existing row to take, which SQLite refuses outright. Give it a default so the existing row backfills to project `1` (the one you created first in Part 3): ```python batch_op.add_column(sa.Column('project_id', sa.Integer(), nullable=False, server_default='1')) ``` (Skip this if your `tasks` table happens to be empty — a fresh scaffold where you skipped straight to Part 4 won't hit it. Either way, apply it:) ```bash zeython db migrate ``` ## The one rule that matters here **Never touch a relationship attribute you didn't explicitly load.** Zeython's ORM is fully async, and an unloaded relationship has no synchronous fallback the way sync SQLAlchemy's lazy loading does — touching one raises `MissingGreenlet`, not a helpful lazy fetch. Load relationships you're about to use with `include=(...)`: ```python task = await Task.find(1, include=("project",)) task.project.name # safe -- already loaded task.to_dict(include=("project",)) # {"id": 1, "title": "...", "project": {"id": 1, "name": "..."}, ...} ``` Update `TaskController` to always eager-load the project, and to accept `project_id` when creating a task: ```python async def index(self, request): tasks = await Task.all(include=("project",)) return JSONResponse([task.to_dict(include=("project",)) for task in tasks]) async def show(self, request): task = await Task.find(int(request.path_params["id"]), include=("project",)) if task is None: raise NotFoundException("Task not found") return JSONResponse(task.to_dict(include=("project",))) async def store(self, request): data = await request.json() task = await Task.create(title=data.get("title"), project_id=data.get("project_id")) task = await Task.find(task.id, include=("project",)) return JSONResponse(task.to_dict(include=("project",)), status_code=201) ``` (`store` re-fetches with `include=` after creating, rather than trying to load the relationship onto the just-saved instance — simpler than juggling two different code paths for "freshly created" vs. "loaded from the database," and one extra indexed lookup is cheap.) Try it: ```bash curl -sS -X POST http://127.0.0.1:8000/tasks \ -H 'Content-Type: application/json' \ -d '{"title": "Design the new homepage", "project_id": 1}' ``` ```json {"title":"Design the new homepage","done":false,"project_id":1,"id":2,"created_at":"...","updated_at":"...","is_deleted":false,"deleted_at":null,"project":{"name":"Website Redesign","id":1,"created_at":"...","updated_at":"...","is_deleted":false,"deleted_at":null}} ``` One request, both objects, no N+1 query for every task in the list. See [Relationships](https://zeython.zaber.dev/docs/relationships/index.md) for the deeper dive — many-to-many, loading a chain of nested relationships, and the dev-only warning that catches a forgotten `include=` before it ships. Continue to [Part 5 — Authentication & Authorization](https://zeython.zaber.dev/docs/tutorial-5-auth/index.md). # Part 5: Authentication & Authorization Right now anyone can create or delete any task. Fix that: creating a task requires being logged in, and only the person who created a task can delete it. Two different questions — *"is anyone logged in"* (authentication) and *"can this specific logged-in user do this specific thing"* (authorization) — and Zeython answers them with two different tools. ## Track who created a task Add an `author` to `app/Models/task.py`, linking to the `User` model that's been there since [Part 1](https://zeython.zaber.dev/docs/tutorial-1-setup/index.md): ```python from sqlalchemy import Boolean, ForeignKey, String from sqlalchemy.orm import Mapped, mapped_column, relationship from zeython import Model, max_length, required from app.Models.project import Project from app.Models.user import User class Task(Model): __tablename__ = "tasks" __rules__ = { "title": [required(), max_length(255)], } title: Mapped[str] = mapped_column(String(255)) done: Mapped[bool] = mapped_column(Boolean, default=False) project_id: Mapped[int] = mapped_column(ForeignKey("projects.id")) author_id: Mapped[int] = mapped_column(ForeignKey("users.id")) project: Mapped[Project] = relationship(back_populates="tasks") author: Mapped[User] = relationship() ``` ```bash zeython db revision -m "add task.author_id" ``` Same real-world migration wrinkle as [Part 4](https://zeython.zaber.dev/docs/tutorial-4-relationships/#migrate): if your `tasks` table already has rows, the generated `author_id` column comes out `nullable=False` with nothing to backfill existing rows with. Give it a default (Ada is user `1`): ```python batch_op.add_column(sa.Column('author_id', sa.Integer(), nullable=False, server_default='1')) ``` ```bash zeython db migrate ``` ## Require login to create a task `require_auth(request)` returns the logged-in user, or raises `UnauthorizedException` (a 401) if nobody's logged in — call it at the top of `store` in `app/Controllers/task_controller.py`: ```python from zeython.auth import require_auth # ... async def store(self, request): user = await require_auth(request) data = await request.json() task = await Task.create(title=data.get("title"), project_id=data.get("project_id"), author_id=user.id) task = await Task.find(task.id, include=("project",)) return JSONResponse(task.to_dict(include=("project",)), status_code=201) ``` ## Only the author may delete their own task `require_auth` alone isn't enough here — *any* logged-in user passes that check, and you specifically want *this task's own creator*. That's what [`Gate`](https://zeython.zaber.dev/docs/authorization/index.md) is for: named abilities, checked against a specific resource. Generate a policy: ```bash zeython make policy Task ``` This created `app/Policies/task_policy.py` with `view`/`create`/`update`/`delete` stubs. Fill in `delete`: ```python class TaskPolicy: def delete(self, user, task) -> bool: return task.author_id == user.id ``` Register it — service providers are where wiring like this lives (see [Part 1](https://zeython.zaber.dev/docs/tutorial-1-setup/index.md) if that's still a fuzzy concept). Create `app/Providers/task_policy_service_provider.py`: ```python from zeython import Gate, ServiceProvider from app.Models.task import Task from app.Policies.task_policy import TaskPolicy class TaskPolicyServiceProvider(ServiceProvider): def boot(self) -> None: gate: Gate = self.container.make(Gate) gate.policy(Task, TaskPolicy) ``` Add it to `main.py`, alongside the `PostPolicyServiceProvider` that's already there: ```python from app.Providers.task_policy_service_provider import TaskPolicyServiceProvider # ... app.register(TaskPolicyServiceProvider(app)) ``` Now enforce it in `destroy`: ```python from zeython.authorization import authorize # ... async def destroy(self, request): task = await Task.find(int(request.path_params["id"])) if task is None: raise NotFoundException("Task not found") await authorize(request, "delete", task) await task.delete() return Response(status_code=204) ``` `authorize()` calls `require_auth` internally (a 401 if nobody's logged in at all), then checks the policy (a 403 if they're logged in but not this task's author). See [Authorization](https://zeython.zaber.dev/docs/authorization/index.md) for `gate.before()` (a global "admins bypass everything" hook) and the rest of the `Gate`/Policy API. ## Try it, including the part that should fail `POST`/`PUT`/`DELETE` to a session-authenticated route need a CSRF header — see [CSRF Protection](https://zeython.zaber.dev/docs/csrf/index.md) for why (the short version: without it, any other website could trigger a `POST` to your app using your visitor's cookie). Extract the token from the cookie jar you've been building since Part 1 and pass it back as a header: ```bash CSRF=$(grep csrf_token cookies.txt | awk '{print $NF}') curl -sS -b cookies.txt -H "X-CSRF-Token: $CSRF" -X POST http://127.0.0.1:8000/tasks \ -H 'Content-Type: application/json' \ -d '{"title": "Ship the redesign", "project_id": 1}' ``` That succeeds — you're logged in as Ada. Now register a second user and try to delete Ada's task as them: ```bash curl -sS -c cookies2.txt -X POST http://127.0.0.1:8000/register \ -H 'Content-Type: application/json' \ -d '{"name": "Bob", "email": "bob@example.com", "password": "hunter2222"}' CSRF2=$(grep csrf_token cookies2.txt | awk '{print $NF}') curl -sS -b cookies2.txt -H "X-CSRF-Token: $CSRF2" -X DELETE http://127.0.0.1:8000/tasks/2 \ -o /dev/null -w '%{http_code}\n' ``` `403` — Bob is logged in (so it's not a 401), but he's not this task's author, so the policy says no. Delete it as Ada instead and it's a clean `204`. Continue to [Part 6 — Testing](https://zeython.zaber.dev/docs/tutorial-6-testing/index.md), where you'll write this exact scenario as a real, repeatable test instead of a sequence of `curl` commands. # Part 6: Testing Five `curl` commands ago you proved Bob can't delete Ada's task. That proof is only good until the next code change quietly breaks it. Turn it into a real test instead, using `zeython.testing` — the same helpers `zeython new` already wired into `tests/test_home.py`. ## How a generated test talks to the app Open `tests/test_home.py` — it's been there since [Part 1](https://zeython.zaber.dev/docs/tutorial-1-setup/index.md): ```python from main import app from zeython.testing import client async def test_index_returns_welcome_message(): async with client(app) as http: response = await http.get("/") assert response.status_code == 200 assert "message" in response.json() ``` `client(app)` gives you an `httpx.AsyncClient` wired directly to your app's ASGI callable — no socket, no `zeython serve` needed. `pyproject.toml` already sets `asyncio_mode = "auto"`, so a plain `async def test_...` function is all pytest needs; no `@pytest.mark.asyncio` decorator. Run it now, before writing anything new, to confirm the baseline works: ```bash pytest ``` ## A fresh database for every test Every generated project also ships `tests/conftest.py`, which points `DATABASE_URL` at `sqlite+aiosqlite:///:memory:` for the whole test run and rebuilds every table before each test function. Without it, `pytest` would share the same `database.db` file `zeython serve` has been writing to all tutorial long — registering `ada@example.com` in a test would collide with the real Ada you registered by hand back in [Part 1](https://zeython.zaber.dev/docs/tutorial-1-setup/index.md). You don't need to touch this file; it's worth knowing it's there so "why does my test see no data from `curl`" never becomes a mystery. ## Test the Task CRUD routes Create `tests/test_tasks.py`. Build every fixture — the project, Ada, Bob — through the same routes `curl` used in Parts 3 and 5, not by poking the database directly; `/register` logs the client in too, so a real request is *less* code than constructing a session by hand: ```python from main import app from zeython.testing import client async def test_creating_a_task_requires_login(): async with client(app) as http: project = (await http.post("/projects", json={"name": "Website Redesign"})).json() response = await http.post("/tasks", json={"title": "Ship it", "project_id": project["id"]}) assert response.status_code == 401 async def test_logged_in_user_can_create_and_see_their_task(): async with client(app) as http: # /register logs the client in too -- see docs/authentication.md. await http.post("/register", json={"name": "Ada", "email": "ada@example.com", "password": "hunter2222"}) project = (await http.post("/projects", json={"name": "Website Redesign"})).json() response = await http.post("/tasks", json={"title": "Ship it", "project_id": project["id"]}) assert response.status_code == 201 body = response.json() assert body["title"] == "Ship it" assert body["project"]["name"] == "Website Redesign" show = await http.get(f"/tasks/{body['id']}") assert show.status_code == 200 assert show.json()["title"] == "Ship it" ``` No `login_as()` needed here — `/register` already sets the session cookie, same as it did for real against `zeython serve` in Part 1. Reach for `login_as()` (see [Testing](https://zeython.zaber.dev/docs/testing/#logging-a-test-client-in-directly)) instead when a test's fixture user already exists some other way and a real register/login round trip would just be noise. ## Test the authorization scenario from Part 5 This is the one that matters — Ada can delete her own task, Bob can't: ```python async def test_only_the_author_can_delete_their_task(): async with client(app) as ada_http: await ada_http.post("/register", json={"name": "Ada", "email": "ada@example.com", "password": "hunter2222"}) project = (await ada_http.post("/projects", json={"name": "Website Redesign"})).json() task = (await ada_http.post("/tasks", json={"title": "Ship it", "project_id": project["id"]})).json() async with client(app) as bob_http: await bob_http.post("/register", json={"name": "Bob", "email": "bob@example.com", "password": "hunter2222"}) response = await bob_http.delete(f"/tasks/{task['id']}") assert response.status_code == 403 response = await ada_http.delete(f"/tasks/{task['id']}") assert response.status_code == 204 ``` Two separate `client(app)` instances, not one client re-logged-in twice — each gets its own cookie jar, so there's no risk of Bob's session leaking onto Ada's requests or vice versa. `403` because Bob *is* authenticated (an anonymous request would get `401`, exactly like `test_creating_a_task_requires_login` above) but the `TaskPolicy` says no; `204` because Ada, the actual author, is allowed. Run it: ```bash pytest tests/test_tasks.py -v ``` ```text tests/test_tasks.py::test_creating_a_task_requires_login PASSED tests/test_tasks.py::test_logged_in_user_can_create_and_see_their_task PASSED tests/test_tasks.py::test_only_the_author_can_delete_their_task PASSED ``` That's the entire Part 5 scenario, pinned down as three tests that run in milliseconds and will fail loudly the moment someone removes the `authorize()` call from `destroy` by accident. See [Testing](https://zeython.zaber.dev/docs/testing/index.md) for `transactional_session` (isolating tests against a real Postgres/MySQL database instead of SQLite's default `:memory:`), factories, and `websocket_client`. ## Where TaskFlow goes from here You've built a real, tested, multi-user application: models with validation, full CRUD controllers, a relationship loaded without an N+1 query, and authorization enforced by a policy instead of scattered `if` checks. Everything you used has a deeper reference page for the cases this tutorial didn't cover: - [Database & Migrations](https://zeython.zaber.dev/docs/database/index.md) — querying beyond `find`/`all`, transactions, raw SQL - [Relationships](https://zeython.zaber.dev/docs/relationships/index.md) — many-to-many, nested `include=` - [Validation](https://zeython.zaber.dev/docs/validation/index.md) — the full built-in rule set, writing your own - [Authorization](https://zeython.zaber.dev/docs/authorization/index.md) — `gate.before()`, ability closures without a policy class - [Testing](https://zeython.zaber.dev/docs/testing/index.md) — factories, transactional isolation, WebSocket tests - [Production Checklist](https://zeython.zaber.dev/docs/production-checklist/index.md) — taking TaskFlow to production Or jump back to the [documentation home](https://zeython.zaber.dev/docs/index.md) and pick whatever's next for what you're building. # The Basics # Database & Migrations ## Defining a model ```python # app/Models/post.py from sqlalchemy import String, Text from sqlalchemy.orm import Mapped, mapped_column from zeython import Model class Post(Model): __tablename__ = "posts" title: Mapped[str] = mapped_column(String(255)) body: Mapped[str] = mapped_column(Text) ``` Every `Model` subclass already has `id`, `created_at`, `updated_at`, `is_deleted`, and `deleted_at` columns — you only declare the columns specific to your table. Register new models in `app/Models/__init__.py` (done automatically by `zeython make model`) so Alembic's autogenerate can see them. ## Active Record API ```python post = await Post.create(title="Hello", body="World") post = await Post.find(1) posts = await Post.all() posts = await Post.find_by(title="Hello") await post.update(title="Updated") await post.delete() # soft delete by default await post.delete(soft=False) # hard delete await post.restore() post.to_dict() # JSON-serializable dict ``` All of these require an active database session — present automatically inside a request, or via `async with database.session():` elsewhere (see [Architecture](https://zeython.zaber.dev/docs/architecture/#request-scoped-database-sessions)). `find`/`all`/`find_by`/`first_where` also accept `include=("relationship_name",)` to eager-load relationships — required reading before you define your first `relationship()`, since touching one without eager-loading crashes async code differently than you'd expect from sync SQLAlchemy. See [Relationships](https://zeython.zaber.dev/docs/relationships/index.md). A model that declares a `tenant_id` column gets every one of these methods scoped to the current tenant automatically — see [Multi-Tenancy](https://zeython.zaber.dev/docs/multi-tenancy/index.md). ## Transactions Every request already runs inside one implicit transaction: `DatabaseSessionMiddleware` opens a session at the start of the request and commits it at the end, or rolls it back if an unhandled exception reaches the end of the request -- even one your own exception handler already turned into a response. Starlette re-raises the original exception to outer ASGI middleware after handling it, specifically so this kind of outer cleanup still runs. Nothing extra is needed for "undo everything this request did if it fails": ```python async def transfer(self, request): await from_account.update(balance=from_account.balance - amount) await to_account.update(balance=to_account.balance + amount) if something_goes_wrong: raise ConflictException("Transfer failed") # both updates above are rolled back -- the whole request's # writes are, whenever an exception ends it ``` `transaction()` is for a narrower case: isolating *part* of a request so a failure there doesn't undo everything else, without ending the request: ```python from zeython import transaction async def checkout(self, request): order = await Order.create(user_id=user.id, status="pending") try: async with transaction(): await reserve_inventory(order) # several writes await charge_payment(order) # might raise except PaymentFailedException: await order.update(status="payment_failed") return JSONResponse({"error": "Payment failed"}, status_code=402) await order.update(status="confirmed") return JSONResponse(order.to_dict()) ``` If `reserve_inventory`/`charge_payment` raise, only their writes roll back (a `SAVEPOINT` under the hood) -- `order`'s initial creation isn't touched, and the handler keeps running to record the failure and respond normally, rather than the whole request dying with a 500. `transaction()` blocks nest: an inner one rolling back doesn't affect an outer one still in progress. Requires an active session, same as the rest of the Active Record API -- raises the same `RuntimeError` as calling `Model.create()` outside one. ## Pagination `all()` loads every matching row — fine for a small table, not for a listing endpoint whose table grows without bound. `paginate()` is the same query, sliced: ```python page = await Post.paginate(page=1, per_page=20) page.items # list[Post] -- this page's rows page.page # 1 page.per_page # 20 page.total # every matching row, not just this page page.total_pages # ceil(total / per_page) page.has_next # page < total_pages page.has_prev # page > 1 ``` `total` costs a second query (a `COUNT(*)` over the same filters) to compute — that's the price of knowing `total_pages`/`has_next` up front, not a bug. If you don't need that, `all()` with a hand-rolled `limit` isn't available on `Model` directly, but nothing stops you from writing a raw `select()` for that one case. `paginate()` accepts the same `include_deleted`/`include=(...)` keywords as `find`/`all`/`find_by`. `zeython new` wires it into the generated `GET /users` (`?page=`/`?per_page=`, defaulting to `1`/`20`) — see `app/Controllers/user_controller.py`. `page.to_dict()` serializes a whole page in one call — items (via each item's own `to_dict()`, for `Model` instances) plus the metadata above: ```python return JSONResponse(page.to_dict()) # {"items": [...], "page": 1, "per_page": 20, "total": 57, # "total_pages": 3, "has_next": true, "has_prev": false} ``` Pass the current request to also get `next_url`/`prev_url` — the same URL with only the `page` query param changed (every other query param carries over), `None` when there is no next/previous page: ```python return JSONResponse(page.to_dict(request=request)) # adds "next_url": "http://.../users?page=2", "prev_url": null ``` ## Connection pooling `DatabaseServiceProvider` forwards `DATABASE_POOL_SIZE`/`DATABASE_MAX_OVERFLOW` straight through to SQLAlchemy's connection pool, unset by default: ```text DATABASE_POOL_SIZE=10 DATABASE_MAX_OVERFLOW=20 ``` - `DATABASE_POOL_SIZE` — steady-state connections the pool keeps open. - `DATABASE_MAX_OVERFLOW` — extra connections allowed beyond that under load, closed again once things quiet down. **Meaningful for PostgreSQL/MySQL, and for a file-based SQLite URL** (what `zeython new` scaffolds, `sqlite+aiosqlite:///./database.db`) — all three default to SQLAlchemy's `AsyncAdaptedQueuePool`, which both settings configure directly. The one exception is `sqlite+aiosqlite:///:memory:` (what the framework's own test suite uses): in-memory SQLite defaults to `StaticPool`, which doesn't accept either kwarg at all — passing them raises `TypeError` at engine-creation time. That's why both are unset by default rather than shipping a number that would break an in-memory setup; set them once you have an actual concurrency figure to size against (a reasonable start: your app server's worker count, or a little above it). ## Read replicas A second `DATABASE_READ_URL` routes read-heavy work to a replica instead of the primary — a report, a dashboard, an analytics query, anything that can tolerate a little replication lag and that you'd rather not have competing with write traffic for the primary's connections: ```text DATABASE_URL=postgresql+asyncpg://user:pass@primary/app DATABASE_READ_URL=postgresql+asyncpg://user:pass@replica/app ``` ```python async def monthly_report(self, request): async with database.read_replica(): orders = await Order.all() return JSONResponse(build_report(orders)) ``` `read_replica()` opens a session against the replica exactly the way `database.session()` opens one against the primary — same `current_session()` underneath, so `Model.find/all/find_by/...` all work inside the block unchanged. Not registered as the request's default session — reach for it explicitly, only for the read path you actually want off the primary; everything else in the request still uses the regular session. **Read-only in practice, not by any check this framework adds.** A real replica is normally configured read-only at the database level itself (Postgres's `default_transaction_read_only`, a MySQL replica user with no write grants) — a write attempted inside `read_replica()` fails with a real database error there, the same as it would against any other client connected to that replica. There's also no `commit()` on exit, since a replica session exists for reads. **Optional** — no `DATABASE_READ_URL` set, `read_replica()` transparently falls back to opening a session against the primary. Code written against it works the same whether or not a replica is actually configured, so it's safe to write `async with database.read_replica():` around a report query in an app that doesn't have one yet. Passed through via `**engine_kwargs` on `Database.__init__` — the same mechanism accepts any other keyword `create_async_engine()` understands, if you construct `Database` yourself instead of going through `DatabaseServiceProvider`. ## Migrations ```bash zeython db revision -m "add posts table" zeython db migrate ``` `zeython new` scaffolds a working Alembic setup (`alembic.ini`, `migrations/env.py`) pointed at your `DATABASE_URL` and your `app.Models` metadata, so `--autogenerate` works out of the box against SQLite, PostgreSQL (`pip install zeython[postgres]`), or MySQL (`pip install zeython[mysql]`). `migrations/env.py` also enables Alembic's `render_as_batch` mode, needed for SQLite specifically: SQLite can add a new column but can't otherwise `ALTER` a constraint in place, so a migration adding a `ForeignKey` (or any other constraint change) to an existing table would fail without it. Harmless no-op on Postgres/MySQL. Adding a `NOT NULL` column to a table that already has rows hits a real SQL constraint on every database, not a Zeython limitation: the existing rows need *some* value for that column. Give the generated migration a default — `sa.Column('author_id', sa.Integer(), nullable=False, server_default='1')` — so the backfill has something to write, or make the column nullable if the data genuinely doesn't apply to old rows. # Relationships Defining a relationship on a `Model` is plain SQLAlchemy — no framework wrapper needed: ```python # app/Models/user.py class User(Model): __tablename__ = "users" posts: Mapped[list["Post"]] = relationship(back_populates="author") # app/Models/post.py class Post(Model): __tablename__ = "posts" author_id: Mapped[int] = mapped_column(ForeignKey("users.id")) author: Mapped[User] = relationship(back_populates="posts") ``` Where Zeython actually adds value is *loading* relationships safely. ## Why this needs care in async code Touching an unloaded relationship in a sync SQLAlchemy app just triggers a lazy load — a bit slow (N+1 queries), but it works. Touching one in an **async** session raises `MissingGreenlet`, because a lazy load is a synchronous DB call and there's no synchronous DB call available inside an async context. This is the single most common way people get stuck the first time they use async SQLAlchemy, and it's exactly what `include=` exists to prevent. ## Eager-loading with `include=` `find`, `all`, `find_by`, and `first_where` all accept `include=(...)`, applying `selectinload()` for each relationship name — the query fetches related rows up front, so touching the relationship afterward (even after the session that fetched it has closed) is just reading already-loaded Python attributes, no further DB access: ```python post = await Post.find(1, include=("author",)) post.author.name # safe -- already loaded posts = await Post.all(include=("author",)) for post in posts: post.author.name # safe for every row -- one extra query total, not N+1 ``` Without `include=`, the same access raises: ```python post = await Post.find(1) post.author.name # MissingGreenlet ``` ## Serializing relationships with `to_dict(include=...)` `to_dict()` accepts the same `include=` and nests the related object's own `to_dict()`: ```python post = await Post.find(1, include=("author",)) post.to_dict(include=("author",)) # {"id": 1, "title": "...", "author": {"id": 1, "name": "Ada", ...}, ...} ``` If you pass `include=` for a relationship that wasn't eager-loaded, `to_dict()` raises a `RuntimeError` telling you exactly what to fix, rather than either silently omitting the data or crashing with `MissingGreenlet` three frames deep in SQLAlchemy: ```python post = await Post.find(1) # no include= post.to_dict(include=("author",)) # RuntimeError: Cannot serialize unloaded relationship 'author' on Post. # Eager-load it first, e.g. `await Post.find(id, include=('author',))`. ``` Without `include=` at all, `to_dict()` only ever touches columns — it never attempts a relationship, loaded or not. ## Assigning a relationship keeps it loaded Setting a relationship attribute directly (rather than just the foreign key column) keeps it loaded in memory immediately, no query needed — useful right after creating a row when you already have the related object: ```python post = await Post.create(title="Hello", body="...", author=current_user) post.to_dict(include=("author",)) # works immediately, no extra query ``` vs. setting only the FK, which leaves the relationship unloaded: ```python post = await Post.create(title="Hello", body="...", author_id=current_user.id) post.to_dict(include=("author",)) # RuntimeError -- author was never loaded ``` ## Scope `include=` loads one level of relationships per name; it doesn't support nested paths like `"comments.author"`. For anything beyond that — complex joins, nested eager loading, custom query shapes — drop down to SQLAlchemy directly with `select()` and `.options()`; `Model` subclasses `Base`, so raw SQLAlchemy queries work on them exactly as documented upstream. ## Detecting N+1s automatically `include=` prevents the crash (`MissingGreenlet`), but forgetting it entirely — hand-rolling a loop that touches a relationship's foreign key and fetches the related row one at a time instead — doesn't crash, it just runs one query per row: fine with 3 rows, a real production slowdown with 3,000. `N1QueryDetectionServiceProvider` catches this in development before it ships: ```python # main.py from zeython import Application, N1QueryDetectionServiceProvider app = Application() app.register(DatabaseServiceProvider) # must run first -- binds Database app.register(N1QueryDetectionServiceProvider(app)) ``` Not registered by default, but its `boot()` is a no-op unless `APP_DEBUG` is true, so it's safe to always register — no per-query overhead and no query text logged from real production traffic. It hooks SQLAlchemy's `before_cursor_execute` event and counts statement *shapes* (bound parameters aside — `SELECT ... WHERE id = ?` run once per row in a loop is one shape, run many times) per request; more than `N1_QUERY_THRESHOLD` (default `10`) of the same shape in one request logs a warning naming the route and the query: ```text WARNING zeython.n_plus_one: Possible N+1 query on /posts: the same statement ran 47 times in one request. Eager-load the relationship instead (include=(...), see docs/relationships.md). Query: SELECT posts.author_id, ... ``` For the full picture of what queries a request ran and how long each took — not just a warning past a threshold — see [Request & Query Profiling](https://zeython.zaber.dev/docs/profiling/index.md). The fix is the same one this whole page is about — eager-load with `include=(...)` instead of touching the relationship inside a loop. # Validation Models validate themselves declaratively via `__rules__`, checked automatically by `save()` (and therefore `create()` and `update()`). ```python from sqlalchemy import String from sqlalchemy.orm import Mapped, mapped_column from zeython import Model, required, email, max_length class User(Model): __tablename__ = "users" __rules__ = { "name": [required()], "email": [required(), email()], "bio": [max_length(500)], } name: Mapped[str] = mapped_column(String(255)) email: Mapped[str] = mapped_column(String(255), unique=True) bio: Mapped[str] = mapped_column(String(500), nullable=True) ``` ```python await User.create(name="Ada", email="not-an-email") # raises zeython.ValidationException({"email": ["Must be a valid email address."]}) ``` `ValidationException` is already handled by the framework's default JSON error handler, so a failed validation in a controller becomes a `422` response with an `errors` object automatically — you don't need a `try`/`except` in most controllers: ```python async def store(self, request): data = await request.json() user = await User.create(**data) # raises -> 422 {"error": ..., "errors": {...}} return JSONResponse(user.to_dict(), status_code=201) ``` ## Available rules | Rule | Behavior | | ------------------ | --------------------------------------------------------------- | | `required()` | Value must not be `None` or `""`. | | `min_length(n)` | String length ≥ `n`. Passes on `None` (pair with `required()`). | | `max_length(n)` | String length ≤ `n`. Passes on `None`. | | `email()` | Basic `local@domain.tld` shape check. | | `one_of((...))` | Value must be one of the given choices. Passes on `None`. | | `matches(pattern)` | Value must match a regex. Passes on `None`. | All rules except `required()` treat `None` as "not applicable" rather than a failure — combine with `required()` when a field is mandatory. ## Checking validity without raising ```python errors = user.validate() # {} if valid, else {"field": ["message", ...]} if errors: ... ``` ## Validating a plain dict `user.validate()` needs a `Model` instance -- not always what you have. A query-string filter, a webhook payload, config loaded from somewhere else: `zeython.validation.validate(data, rules)` runs the same rule sets against any dict: ```python from zeython.validation import validate, required, email errors = validate( {"email": "not-an-email"}, {"name": [required()], "email": [required(), email()]}, ) # {"name": ["This field is required."], "email": ["Must be a valid email address."]} if errors: raise ValidationException(errors) ``` A missing key is treated as `None`, same as an unset field on a model instance. `Model.validate()` is this function applied to a model's own field values -- the two always agree on what the same rule set means. ## Custom rules A rule is just `Callable[[Any], bool]` wrapped with a message: ```python from zeython import Rule def even(message="Must be even.") -> Rule: return Rule(lambda v: v is None or v % 2 == 0, message) ``` # Model Events Every `Model` has overridable lifecycle hooks — no-ops by default, and the right place to react to (or shape) a save or delete without cluttering the controller that triggered it. ## The hooks `save()` (which `create()` and `update()` both go through) calls, in order: ```text saving() -> creating() or updating() -> [write to the database] -> created() or updated() -> saved() ``` `delete()` calls `deleting()` before removing/soft-deleting the row, then `deleted()` after: ```python class User(Model): ... async def saving(self) -> None: # Runs before both create and update. self.email = self.email.strip().lower() async def creating(self) -> None: # Runs only on the first save. ... async def deleted(self) -> None: # Runs after delete() -- soft or hard. ... ``` Override only the hooks you need; the rest stay no-ops. ## Hooks run before validation `creating()`/`updating()` run **before** `__rules__` is checked — this is what makes them useful for deriving a field validation then depends on, not just for reacting after the fact: ```python class Post(Model): __rules__ = {"slug": [required()]} title: Mapped[str] = mapped_column(String(255)) slug: Mapped[str] = mapped_column(String(255), default="") async def creating(self) -> None: if not self.slug: self.slug = self.title.lower().replace(" ", "-") ``` `await Post.create(title="Hello World")` succeeds — `slug` is derived before `__rules__` ever sees it. If the ordering were reversed, this would raise `ValidationException` on an empty `slug` every time. ## Distinguishing create from update `is_new` isn't exposed as a hook argument — `creating()`/`created()` firing at all *is* that signal; `updating()`/`updated()` fire on every subsequent save. If a hook genuinely needs both cases in one method, use `saving()`/ `saved()`, which fire on every save regardless. ## Observers A model's own hooks are one implementation per class — fine for behavior that belongs to the model itself (deriving a slug, normalizing an email). For a cross-cutting concern that doesn't belong on the model, or that several independent things want to react to (search-index sync, cache invalidation, audit logging), register an `Observer` instead: ```python from zeython import Observer class PostSearchIndexObserver(Observer): async def created(self, model: Post) -> None: await search_index.add(model.id, model.title) async def updated(self, model: Post) -> None: await search_index.update(model.id, model.title) async def deleted(self, model: Post) -> None: await search_index.remove(model.id) Post.observe(PostSearchIndexObserver) ``` `observe()` accepts a class (instantiated with no arguments) or an already-constructed instance — typically called once, e.g. in a service provider's `boot()`. An observer has the same eight hooks as a model (`saving`/`saved`/`creating`/`created`/`updating`/`updated`/`deleting`/ `deleted`), each taking the model instance as its argument; override only the ones you need. Several observers can watch the same model, and each model class's observers are independent of every other model's. Observers fire after the model's own same-named hook, in the order shown above. ## What this isn't An observer doesn't replace a model's own hooks — it's for reactions that don't belong on the model, not a place to move logic that does. If a hook is really about the model's own data (deriving a field, normalizing input before validation), keep it a method on the model, not an observer. It's also not for application-defined events that aren't tied to a specific model's lifecycle at all (`OrderPlaced`, a scheduled report finishing) — see [Events](https://zeython.zaber.dev/docs/events/index.md) for those. # Database Factories & Seeders A fresh `zeython db migrate` gives you empty tables. Getting from there to something a developer can actually click around in -- an admin user, a handful of demo posts -- or a fixed reference row production genuinely needs, is what factories and seeders are for. ## Factories: building model instances A factory defines default attributes for one model, so tests and seeders don't need to spell out every column by hand: ```bash zeython make factory Post ``` ```python # database/factories/post_factory.py from zeython import Factory from app.Models.post import Post class PostFactory(Factory[Post]): model = Post def definition(self, sequence: int) -> dict: return { "title": f"Post {sequence}", "body": f"This is the body of post {sequence}.", } ``` ```python post = await PostFactory().create() # built and saved posts = await PostFactory().create_many(5) # five saved rows draft = PostFactory().make(title="Not saved yet") # built, not persisted ``` `definition()` receives a `sequence` number that starts at 1 and increments on every call against that factory instance -- use it to keep unique-constrained columns (an email, a slug) actually unique across a batch: ```python def definition(self, sequence: int) -> dict: return {"email": f"user{sequence}@example.com"} ``` There's no bundled fake-data library. For simple cases, an f-string built from `sequence` is often enough (as above); for realistic names, addresses, or similar, add [Faker](https://faker.readthedocs.io/) yourself and call it inside `definition()` -- it's a plain Python method, nothing framework-specific to wire up. ### Overrides and relationships Keyword arguments to `make()`/`create()` override `definition()`'s defaults. This is also how a `belongs_to` relationship's foreign key usually gets set -- a factory can't call another factory's `create()` from inside `definition()` (`definition()` is synchronous; `create()` isn't), so pass it explicitly instead: ```python user = await UserFactory().create() post = await PostFactory().create(author_id=user.id) ``` ### `create()` needs an active session `create()`/`create_many()` call `Model.create()` under the hood, so they need the same active database session every other persistence method does -- a request, or an explicit `async with database.session():` block (which is exactly what `zeython db seed` opens for you; see below). ## Seeders: populating the database ```bash zeython make seeder User ``` ```python # database/seeders/user_seeder.py from zeython import Seeder from database.factories.user_factory import UserFactory class UserSeeder(Seeder): async def run(self) -> None: await UserFactory().create(email="admin@example.com") await UserFactory().create_many(9) ``` `self.app`/`self.container` are available inside `run()`, same as a `Command`. Compose multiple seeders from one entry point with `self.call(...)`: ```python # database/seeders/database_seeder.py from zeython import Seeder from database.seeders.post_seeder import PostSeeder from database.seeders.user_seeder import UserSeeder class DatabaseSeeder(Seeder): async def run(self) -> None: await self.call(UserSeeder, PostSeeder) ``` `DatabaseSeeder` is the conventional entry point -- a generated project ships one, seeding a handful of demo users and posts via `database/factories/`. ## Running seeders ```bash zeython db seed # runs DatabaseSeeder zeython db seed --class UserSeeder # runs a specific seeder instead ``` `zeython db seed` opens one database session for the whole run (so every `Model.create()`/factory `create()` call inside just works, no session plumbing needed), runs the seeder, and commits. Tables must already exist -- run `zeython db migrate` first. There's no `--force`/environment guard: nothing stops you from running a seeder against a production database. Keep destructive or non-idempotent seed data (a `TRUNCATE`, unconditionally inserting the same admin user) out of a seeder you'd run more than once, the same way you would with a migration. # Views Zeython renders server-side HTML with Jinja2, by convention reading templates from `resources/views/`. ```python # app/Controllers/page_controller.py from zeython import Controller from zeython.views import render class PageController(Controller): async def show(self, request): return render(request, "pages/show.html", {"title": "Hello"}) ``` ```html {{ title }}

{{ title }}

``` ## Wiring it up `render()` looks up the application's `Views` instance from the container, so register `ViewServiceProvider` alongside your other providers: ```python # main.py from zeython import Application, DatabaseServiceProvider, RouteServiceProvider, ViewServiceProvider app = Application() app.register(DatabaseServiceProvider) app.register(ViewServiceProvider) app.register(RouteServiceProvider(app, modules=("routes.web",))) ``` ## Configuration By default templates are read from `/resources/views`. Override with `VIEWS_PATH` in `.env`: ```text VIEWS_PATH=templates ``` ## Using the `Views` object directly `render()` is sugar over the container-bound `Views` instance; you can also resolve and use it directly if you need more control (custom filters, globals): ```python from zeython import Views views = app.container.make(Views) views.environment.filters["shout"] = lambda s: s.upper() ``` # Frontend & CSS Zeython is a backend framework — it renders HTML via Jinja2 (`resources/views/`, see [Views](https://zeython.zaber.dev/docs/views/index.md)) and leaves the frontend build tooling up to you. It doesn't bundle a JS framework, a Vite pipeline, or a Node.js dependency of any kind. What it does ship an opinion on is CSS, because "unstyled HTML" is a bad first impression for a new project. ## Tailwind out of the box (dev only) The generated `resources/views/welcome.html` loads Tailwind's **Play CDN**: ```html ``` This gives every new project real, non-ugly styling with zero setup — no `npm install`, no build step, no Node.js requirement at all. Use utility classes in any `.html` template and they just work. **This is not a production setup.** The Play CDN compiles every utility class in the browser, on every page load, with nothing purged — Tailwind's own docs are explicit that it's for prototyping, not deployment. Before you ship, replace it with a compiled build. ## Moving to a compiled build You do not need Node.js to compile Tailwind — the standalone CLI is a single executable with no runtime dependency: ```bash # macOS/Linux, see https://tailwindcss.com/blog/standalone-cli for other platforms curl -sLO https://github.com/tailwindlabs/tailwindcss/releases/latest/download/tailwindcss-linux-x64 chmod +x tailwindcss-linux-x64 mv tailwindcss-linux-x64 tailwindcss ./tailwindcss -i resources/css/app.css -o public/css/app.css --minify ``` Then swap the CDN `