Deploying it¶
A deployment is a sequence, and most of what goes wrong is the sequence rather than the code. This page is the sequence: build the application from configuration, check it, migrate the database, serve it, and answer readiness honestly — with the two choices that are actually choices marked as such.
The application is examples/learn_production/deploy01_app.py. It is small on purpose: two routes,
one service, one database. What makes it deployable is not its size.
The factory contract¶
examples/learn_production/deploy01_app.py
"""The application a deployment runs: eager database, a service, and a readiness route.
Nothing here is production-only machinery, and that is the point. What makes this deployable is
that the factory takes no arguments, the database is opened at *startup* rather than on the first
request, and readiness is a route the deployment asks rather than a property the process assumes.
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from contextlib import AbstractAsyncContextManager
from pathlib import Path
from typing import Annotated, TypeVar
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncConnection
from starlette.applications import Starlette
from xyberos import App, Inject, Settings
from xyberos.database import (
DatabaseProvider,
SqliteSettings,
TransactionOperation,
open_database,
sqlite_provider,
)
from xyberos.errors import DatabaseConnectionError, HttpError, XyberosError
from xyberos.http import HttpAdapter
#: The configuration file this application ships, read before the environment overrides it.
SETTINGS_FILE = Path(__file__).with_name("deploy.toml")
#: The one setting that is not the runtime's business, so the application reads it itself.
DATABASE_KEY = "ORDERS_DATABASE"
T = TypeVar("T")
class NotReadyError(HttpError):
"""A readiness probe that failed, in the shape a load balancer understands."""
def __init__(self) -> None:
super().__init__("the database is not reachable", status_code=503)
class OrderNotFoundError(HttpError):
"""A client mistake, and the only thing this application says about a missing row."""
def __init__(self, order_id: int) -> None:
super().__init__(f"order {order_id} does not exist", status_code=404)
class UnavailableDatabase:
"""A provider that never opened, so readiness can report it instead of a crash at startup.
This is the difference between the two deployment postures the walkthrough compares. A handler's
injected dependencies are resolved *before* its body runs, so a provider whose construction
raises cannot be caught by the route that wanted to report the failure: the request becomes a
500 with a message that says nothing. Registering this instead keeps the failure reportable —
every call raises the original ``DatabaseConnectionError``, and `/health` turns it into a 503.
"""
def __init__(self, error: XyberosError) -> None:
self._error = error
def transaction(self) -> AbstractAsyncContextManager[AsyncConnection]:
"""Refuse, the way a provider that never opened must."""
raise self._error
def retrying(
self,
operation: TransactionOperation[T],
*,
attempts: int | None = None,
base_delay: float | None = None,
) -> Awaitable[T]:
"""Refuse on await, so the caller sees the failure rather than a missing method."""
async def fail() -> T:
raise self._error
return fail()
async def health(self) -> None:
"""Report the failure this provider was built with."""
raise self._error
async def aclose(self) -> None:
"""Nothing to release: nothing was opened."""
class OrderService:
"""Reads orders through the contract, so the engine behind it is somebody else's decision."""
def __init__(self, database: Annotated[DatabaseProvider, Inject()]) -> None:
self._database = database
async def total(self, order_id: int) -> float:
"""Return one order's total, retrying write contention, or report that it is not there."""
async def read(connection: AsyncConnection) -> float | None:
result = await connection.execute(
text("SELECT total FROM orders WHERE id = :id"), {"id": order_id}
)
value = result.scalar_one_or_none()
return None if value is None else float(value)
total = await self._database.retrying(read)
if total is None:
raise OrderNotFoundError(order_id)
return total
def tolerant_provider(settings: SqliteSettings) -> Callable[[], Awaitable[DatabaseProvider]]:
"""Return a factory whose database failure is a reportable state rather than a start failure."""
async def open_tolerantly() -> DatabaseProvider:
try:
return await open_database(settings)
except DatabaseConnectionError as error:
return UnavailableDatabase(error)
return open_tolerantly
def create_app(environ: dict[str, str] | None = None, *, eager: bool = True) -> App:
"""Build the application from configuration, with no *required* arguments.
That last property is a deployment contract rather than a style choice: `xyberos check` and an
ASGI server both call this factory by name, so one that needed an argument could not be named.
``eager`` is the deployment's posture towards a broken database, and the two are not equivalent:
* ``eager=True`` — the database is opened at startup, so a bad path is a failure to *start*. The
orchestrator sees a crash loop, which is what it is built to report.
* ``eager=False`` — the process comes up and `/health` answers ``503`` until the database is
reachable, which keeps a rolling deployment from taking every replica down at once.
"""
overrides = environ if environ is not None else {}
settings = Settings.load(config_path=SETTINGS_FILE, environ=overrides)
path = Path(overrides.get(DATABASE_KEY, "orders.db"))
database = SqliteSettings(path=path)
app = App(settings=settings)
if eager:
app.provide(DatabaseProvider, sqlite_provider(database), eager=True)
else:
app.provide(DatabaseProvider, tolerant_provider(database))
app.provide(OrderService)
app.get("/orders/{order_id}")(read_order)
app.get("/health")(health)
return app
async def read_order(order_id: int, orders: Annotated[OrderService, Inject()]) -> dict[str, object]:
"""Answer with one order's total, or with the 404 the service raises."""
return {"order": order_id, "total": await orders.total(order_id)}
async def health(database: Annotated[DatabaseProvider, Inject()]) -> dict[str, object]:
"""Report readiness by asking the database, not by reporting that the process is running.
A process that answers ``200`` while it cannot reach its database is the failure mode readiness
checks exist to prevent, so this route asks the provider and turns a refusal into a ``503`` the
load balancer understands — with a message that is safe to expose because it says nothing about
why. The provider the handler receives is the application's own, never a second connection.
"""
try:
await database.health()
except DatabaseConnectionError:
raise NotReadyError() from None
return {"ready": True}
def create_asgi(*, eager: bool = True) -> Starlette:
"""Return the ASGI application a server runs, compiled from the runtime built above.
Two entry points rather than one, because they answer different questions: an ASGI server needs
something to serve, and `xyberos check` needs the runtime to inspect *without* serving it. The
adapter compiles the routes into a Starlette application; the ``App`` behind it is the runtime,
and it is the same object either way.
"""
return HttpAdapter(create_app(eager=eager)).asgi_app
Three properties carry the weight, and all three are checkable rather than aspirational:
create_apptakes no required arguments. Bothxyberos checkand a server call a factory by name, so a factory that needed an argument could not be named in the first place. Everything the application needs arrives fromSettings.load, which reads the file below and then the environment.create_asgiis the second entry point, and it exists for a different caller. A server needs something to serve, socreate_asgicompiles the same runtime into a Starlette application throughHttpAdapter;xyberos checkneeds the runtime itself, so it namescreate_app. One factory and two names beats one factory that guesses who is calling.- The database is opened at startup, not on the first request. That is what
eager=Truedoes, and it changes what a broken database is: a failed start rather than a failed request. Theeager=Falsevariant further down is the other posture, and the difference between them is a deployment decision with an operational consequence.
Configuration, in the order it is read¶
examples/learn_production/deploy.toml
[app]
name = "xyberos"
environment = "development"
debug = false
The file states the deployment's defaults; the environment overrides them, which is how the same artifact runs in two places without being rebuilt:
| Source | Example | Precedence |
|---|---|---|
explicit overrides passed to Settings.load |
environ={"ORDERS_DATABASE": ":memory:"} |
highest |
| environment variables | XYBEROS_APP__DEBUG=true, XYBEROS_CONFIG=/etc/app.toml |
above the file |
| the TOML file | deploy.toml above |
above the defaults |
| typed defaults | app.environment = "development", app.debug = false |
lowest |
ORDERS_DATABASE in that table is not a runtime setting: it is the one thing this application reads
itself, with a name of its own, because "where the orders live" is the application's business rather
than the framework's. Nested settings use __, and a key the runtime does not know is refused rather
than ignored — security shows what that refusal prints, which is the setting's name and
never its value. The whole configuration surface is in
the runtime reference.
Check it before you run it¶
check composes the application without starting it, then inspects what it composed: the provider
graph, the plugins, and every route's handler signature. It is the cheapest failure you can have,
because it happens before a server binds a port. The same call, in-process:
examples/learn_production/deploy02_check.py
"""Walkthrough — the preflight a deployment runs before it serves anything.
Run it with:
python -m learn_production.deploy02_check
"""
from __future__ import annotations
from xyberos.cli.main import main as cli
#: The factory a deployment names, in the `module:attribute` form the CLI and an ASGI server share.
TARGET = "learn_production.deploy01_app:create_app"
def main() -> None:
"""Check the application offline, and report what the check decided."""
code = cli(["check", "--app", TARGET, "--app-dir", "examples"])
print("xyberos check exited with", code)
if __name__ == "__main__":
main()
settings: xyberos (development), schema_version 1, debug False
application: learn_production.deploy01_app:create_app
provider graph: valid
providers:
DatabaseProvider [app, eager] from application
OrderService [app] from application
plugins:
(none)
routes: 2 (handler signatures valid)
xyberos check exited with 0
Five lines, five questions answered before the process serves anything:
| Line | What a deployment learns |
|---|---|
settings: xyberos (development), schema_version 1, debug False |
which configuration was actually loaded, including the schema version the runtime understands — the two things that differ first when two deployments disagree |
application: ...:create_app |
the factory the CLI resolved, so a typo is a reported unresolvable target rather than a crash loop |
provider graph: valid |
every declared dependency has a provider, no cycle, no scope violation — the checks troubleshooting prints one at a time |
DatabaseProvider [app, eager] from application |
the scope, whether it is eager, and who registered it: the source field that makes an unexpected override traceable |
routes: 2 (handler signatures valid) |
the handler signatures the adapter will inject into are valid, which is a different question from the graph being valid |
The environment on that line is worth dwelling on: it is how the same artifact says which
deployment it believes it is in, and a staging process reporting production is one line away from
being spotted. The command's own reference is the CLI guide.
Migrate before you serve¶
The deploy step that decides whether the first request works:
python -m xyberos migrate --url "sqlite:///var/orders.db" --migrations examples/learn_production/migrations
Nothing migrates at startup, and that is deliberate: every replica would try, on every boot, and a slow migration would be a container that never becomes healthy. The application ships the migrations next to the tables they create:
examples/learn_production/orders_tables.py
"""The tables this application owns, as SQLAlchemy Core metadata.
The application queries these tables through the database provider's transactions; the migration
below creates them. Both read the same definition of the table names and columns, which is the part
worth keeping in one place even when the DDL is written out.
"""
from __future__ import annotations
from sqlalchemy import Column, Float, Integer, MetaData, Table
metadata = MetaData()
orders = Table(
"orders",
metadata,
Column("id", Integer, primary_key=True, autoincrement=True),
Column("total", Float, nullable=False),
)
examples/learn_production/migrations/versions/0001_create_orders.py
"""create the orders table
Revision ID: 0001_create_orders
Revises:
Create Date: 2026-10-12
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
revision = "0001_create_orders"
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"orders",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column("total", sa.Float(), nullable=False),
)
def downgrade() -> None:
op.drop_table("orders")
The revision imports the same metadata object the application's models use, so a schema drift between
the two is not possible to write by accident. The env.py that Alembic runs is where the one real
gotcha lives:
examples/learn_production/migrations/env.py
"""Alembic environment for the orders application.
The database URL arrives from the caller — application code, or the `xyberos migrate` command — so
no URL is committed here. The application's ``metadata`` is the target for autogenerate, which is
why the package is made importable before it is read.
Alembic's own engine is synchronous, so the URL passed in is the synchronous form of the one the
application uses (``SqliteSettings.sync_url`` for SQLite).
"""
from __future__ import annotations
import sys
from logging.config import fileConfig
from pathlib import Path
from alembic import context
from sqlalchemy import engine_from_config, pool
# Make the application importable when Alembic runs from any working directory.
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from learn_production.orders_tables import metadata # noqa: E402
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = metadata
def run_migrations_offline() -> None:
"""Emit SQL for the migration without connecting to a database."""
context.configure(
url=config.get_main_option("sqlalchemy.url"),
target_metadata=target_metadata,
literal_binds=True,
render_as_batch=True,
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""Apply the migration against a live database."""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
render_as_batch=True,
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
Alembic's engine is synchronous, so it needs the synchronous URL. SqliteSettings keeps both forms
for the same file: url is sqlite+aiosqlite:// for the application, sync_url is sqlite:// for the
migration. Passing the async URL to apply_migrations is the mistake this page exists to prevent, and
it fails loudly rather than halfway. The proof, run against a temporary database:
examples/learn_production/deploy03_migrations.py
"""Walkthrough — migrations as a deployment step, and how to read where one landed.
Run it with:
python -m learn_production.deploy03_migrations
"""
from __future__ import annotations
import asyncio
import tempfile
from pathlib import Path
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from xyberos.database import SqliteSettings, apply_migrations
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
#: The migrations this application ships, next to the tables they create. A deployment runs the same
#: command against this directory: `xyberos migrate --migrations ... --url ...`.
MIGRATIONS = Path(__file__).with_name("migrations")
async def apply_to(path: Path) -> None:
"""Apply every revision to the database file at ``path``, the way `xyberos migrate` does.
Alembic's own engine is synchronous, so the URL is the *synchronous* one — which is what
`SqliteSettings.sync_url` is for. The application keeps using the async URL for the same file.
"""
await apply_migrations(MIGRATIONS, url=SqliteSettings(path=path).sync_url)
async def main() -> None:
"""Migrate a database, then ask the database itself where it landed."""
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "orders.db"
await apply_to(path)
engine = create_async_engine(SqliteSettings(path=path).url)
try:
async with engine.connect() as connection:
version = await connection.execute(text("SELECT version_num FROM alembic_version"))
tables = await connection.execute(
text("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name")
)
print("schema version:", version.scalar_one())
print("tables:", ", ".join(row[0] for row in tables))
finally:
await engine.dispose()
if __name__ == "__main__":
asyncio.run(main())
The check asks the database where it landed rather than trusting the migration command's exit code:
alembic_version holds the revision, and orders either exists or does not. A deployment's own smoke
test should do the same thing — a migration process that reported success and a schema that is not
there is exactly the failure a status code cannot see.
Refusing to start, or reporting not-ready¶
This is the one genuine choice on the page, and the two options are not equivalent:
| Posture | What happens with an unreachable database | When it is right |
|---|---|---|
eager=True (the default) |
the process fails to start; the orchestrator sees a crash loop | a fresh deployment with no traffic to lose, and a database that should be there |
eager=False |
the process starts, /health answers 503 until the database answers |
a rolling deployment, where a crash loop would take every replica down together |
Both, in one transcript:
examples/learn_production/deploy04_readiness.py
"""Walkthrough — readiness, and the choice between failing fast and reporting the truth.
Run it with:
python -m learn_production.deploy04_readiness
"""
from __future__ import annotations
import asyncio
import tempfile
from pathlib import Path
from typing import cast
from sqlalchemy import text
from learn_production.deploy01_app import DATABASE_KEY, create_app
from learn_production.deploy03_migrations import apply_to
from xyberos import App
from xyberos.database import DatabaseProvider
from xyberos.errors import XyberosError
from xyberos.http import HttpAdapter
from xyberos.http.testing import create_test_client
async def seed(app: App, total: float) -> None:
"""Insert the one row the walkthrough reads, through the application's own provider."""
async with app.lifespan():
# `resolve` returns `T` for a class and `object` for its contract overload; mypy does not
# match the first against a Protocol class, so the runtime type is spelled out here.
database = cast(DatabaseProvider, await app.resolve(DatabaseProvider))
async with database.transaction() as connection:
await connection.execute(
text("INSERT INTO orders (id, total) VALUES (1, :total)"), {"total": total}
)
async def scenario() -> None:
"""Ask the same application about an unreachable database, then about a real one."""
with tempfile.TemporaryDirectory() as directory:
# A directory is a path no SQLite file can be, which is the cheapest unreachable database.
unreachable = str(Path(directory) / "orders.db")
Path(unreachable).mkdir()
patient = create_app({DATABASE_KEY: unreachable}, eager=False)
with create_test_client(HttpAdapter(patient)) as client:
refused = client.get("/health")
print("eager=False ->", refused.status_code, refused.json())
try:
eager = create_app({DATABASE_KEY: unreachable})
async with eager.lifespan():
pass
except XyberosError as error:
cause = type(error.__cause__).__name__ if error.__cause__ else "none"
print("eager=True ->", type(error).__name__, "caused by", cause)
print("and the application is left in state:", eager.state.value)
path = Path(directory) / "migrated.db"
await apply_to(path)
await seed(create_app({DATABASE_KEY: str(path)}), 42.5)
healthy = create_app({DATABASE_KEY: str(path)})
with create_test_client(HttpAdapter(healthy)) as client:
print("readiness ->", client.get("/health").json())
print("GET /orders/1 ->", client.get("/orders/1").json())
missing = client.get("/orders/7")
print("GET /orders/7 ->", missing.status_code, missing.json())
def main() -> None:
"""Run the scenario."""
asyncio.run(scenario())
if __name__ == "__main__":
main()
eager=False -> 503 {'error': {'type': 'NotReadyError', 'message': 'the database is not reachable'}}
eager=True -> DependencyResolutionError caused by DatabaseConnectionError
and the application is left in state: failed
readiness -> {'ready': True}
GET /orders/1 -> {'order': 1, 'total': 42.5}
GET /orders/7 -> 404 {'error': {'type': 'OrderNotFoundError', 'message': 'order 7 does not exist'}}
Four things about those six lines:
503with a typed body. Readiness is a route that asks the database, and its failure is anHttpErrorsubclass, so the client gets the status a load balancer understands and a message that says nothing about why — no path, no driver, no connection string.DependencyResolutionError caused by DatabaseConnectionErrorandstate: failed. Under the other posture the same database is a failed start, and the state is observable afterwards rather than guessed at. Internals walks the states.- A handler's dependencies are resolved before its body runs. This is why the tolerant
UnavailableDatabaseprovider exists in the module: a provider whose construction raises cannot be caught by a route, because the route never runs. If you want to report a broken dependency instead of refusing to start, the dependency has to be able to construct and fail on use. 404is a route's own decision.OrderNotFoundErroris anHttpErrorwith a status code, so the adapter maps it; anything that is not anHttpErrorbecomes a generic500and keeps its message in the log (security shows both halves of that in one transcript).
The deploy sequence¶
python -m pip install "xyberos[http,database]" uvicorn
python -m xyberos check --app learn_production.deploy01_app:create_app --app-dir examples
python -m xyberos migrate --url "sqlite:///var/orders.db" --migrations examples/learn_production/migrations
uvicorn --app-dir examples --factory learn_production.deploy01_app:create_asgi
curl http://127.0.0.1:8000/health
Five steps, and the order is the content: install, check the graph, migrate, serve, then let the readiness endpoint decide whether this process joins the rotation.
What the deployment still owns¶
| Job | What to do | What Xyberos does |
|---|---|---|
| TLS, CORS, body and rate limits | at the server and the proxy | nothing, on purpose (security reference) |
| Secrets | environment or a secret manager, never the image | reads the environment; does not fetch, rotate, or audit |
| Identity | the proxy authenticates and strips its own assertion headers | nothing: the traditional path's identity lesson shows the application-side shape |
| Backups | the database file and its WAL while the process runs | nothing |
| Shutdown | send SIGTERM and wait for the drain |
stops accepting new work, drains in-flight request contexts, then closes in reverse order (internals). Set app.shutdown_timeout to bound the drain: None, the default, waits as long as it takes |
| Log shipping | configure a root handler | binds a correlation id per request and logs nothing to a sink of its own (observability) |
Common mistakes¶
- Serving before migrating. The first request that touches the new column is a
500, and the deploy that "worked" is the one that never exercised it. - A readiness check that only proves the process is alive.
{"ready": true}is a claim about the database, not about the event loop; the route above asks the provider for that reason. --factorypointing at the runtime factory instead of the ASGI one.create_appreturns anApp, which is not an ASGI application;create_asgiis the one a server can call.- Passing the async URL to Alembic. See
sync_urlabove: the failure is loud, but only if you look. debug=Truein production. It changes how internal errors are reported, and it is printed byxyberos check, which is one command.- Restarting a stopped application instead of replacing the process.
Appdoes not start again once stopped, by design; the orchestrator's job is to start a new process. Internals shows the refusal.