Runtime
Application
App extends Router with the app entrypoint, error surface, shutdown-aware scheduling, and reverse URLs. The CLI or Server constructs the app; bootstrap(app, span) receives that instance—avoid instantiating App() yourself in normal apps (see Routing for the same pattern). Listener binding and signal handling are documented under Server on this page.
On this page you will find: lifecycle and bootstrap shapes, exception handling and on_error, shutdown-aware tasks, url_for, and static assets below.
Application lifecycle
Bootstrap return shapes at a glance:
async def bootstrap(app, span) -> None— awaited once; no framework teardown after it returns (unless you use shutdown hooks elsewhere).def bootstrap(app, span) -> None— synchronous wiring when you do not needawait.async def bootstrap(app, span) -> AsyncIteratorwith exactly oneyield— code beforeyieldis startup; code after runs on shutdown.Other shapes are normalized via
normalize_bootstrap(async context managers, etc.); see the apidoc under Server.
At process level: bootstrap (wire the app before listening), serving, and shutdown. Every bootstrap(app, span) goes through normalize_bootstrap, whether you use Server, stario.testing, or the stario console entry point. Supported return shapes differ in whether a second teardown phase runs after shutdown begins. The normalize_bootstrap apidoc under Server lists every accepted shape, including async context managers.
A synchronous def bootstrap(app: App, span: Span) -> None is supported (no await in the body)—useful for quick wiring when everything you need is synchronous; prefer async def when you need to await setup.
Use span.attr, span.attrs, and span.event during bootstrap while mounting routes or opening resources. The same span object is reused for the whole Server.run() scope, but the startup phase completes with span.end() once the server is running; after that, shutdown logic may repoint span.id to a shutdown span so late telemetry still lands in the right place. See Server.run in the framework source for the exact sequence.
from collections.abc import AsyncIteratorfrom pathlib import Path from stario import App, Context, Span, StaticAssets, Writer async def home(c: Context, w: Writer) -> None: ... async def bootstrap(app: App, span: Span) -> None: span.attrs({"bootstrap": "await-once"}) app.mount("/static", StaticAssets(Path("static"), name="static")) app.get("/", home, name="home") async def bootstrap_with_teardown(app: App, span: Span) -> AsyncIterator[None]: span.attr("phase", "startup") # Open DB pools, clients, … yield span.attr("phase", "shutdown") # Close resourcesPoint stario serve main:bootstrap (or Server) at whichever function is your real entry (bootstrap or bootstrap_with_teardown above—only one should be the module’s bootstrap export).
Graceful shutdown (typically SIGINT/SIGTERM) stops accepting connections, waits up to graceful_timeout for open connections and for tasks from app.create_task, then closes transports, cancels stragglers, and tears down the listener. Adjust the window on Server (default five seconds). For long request work such as streams or SSE, prefer Writer.alive(). Work that must outlive that wave belongs outside the per-request path or needs an explicit longer grace.
After bootstrap, each inbound message runs App.__call__(context, writer): request span, route resolution, errors, and always writer.end() for keep-alive. Handlers use context and writer, not the bootstrap span.
Exception handling
When a handler or middleware lets an exception propagate, App resolves an on_error handler by walking the exception’s MRO; the most specific registered type wins.
HttpException is the normal way to turn a failed precondition into an HTTP response body (404 text, 400 message, and so on). For redirects (3xx), prefer RedirectException so the Location value is distinct from a body string: from stario import RedirectException, e.g. raise RedirectException(307, "/login"). Both inherit the same default on_error handler (exc.respond(w)). Import from stario import HttpException for non-redirect outcomes (re-exported from stario.exceptions). Replace the default handler only if you want different behavior for every intentional HTTP outcome.
Other errors become a response only while the response has not started (writer.started is false). After the status line and headers are on the wire, the runtime records the failure on the request span and ends the writer; it cannot send a late 5xx or change headers for that exchange.
Use app.on_error for domain or infrastructure types you want mapped to stable responses in one place instead of repeating branches in every handler. Practical patterns: Mapping errors to HTTP responses (how-to) and this Exception handling section.
from stario import App, Context, Span, Writerimport stario.responses as responses class OrderNotFound(Exception): pass def order_not_found(c: Context, w: Writer, exc: OrderNotFound) -> None: responses.text(w, "Unknown order", status=404) async def bootstrap(app: App, span: Span) -> None: app.on_error(OrderNotFound, order_not_found)Shutdown-aware tasks
app.create_task(coro) schedules the coroutine and adds the resulting asyncio.Task to a set that Server tracks during shutdown. That set shares graceful_timeout with open connections. After the window, remaining tasks are cancelled and awaited with gather(..., return_exceptions=True).
Bare asyncio.create_task is not in that set and is easy to orphan on shutdown. Prefer app.create_task for work that should share the server’s graceful window (bootstrap loops, request-adjacent work). The HTTP stack uses this path for dispatch and pipelining.
Reverse URLs
url_for builds a path (and optional query) from the name= on a route or static mount. It matters most when the public URL is not fixed at authoring time, especially fingerprinted static files. Without name= on a mount, there is no reverse entry. Mount and collision details: Routing, Static assets.
from pathlib import Path from stario import App, Context, Span, StaticAssets, Writer async def home(c: Context, w: Writer) -> None: ... async def bootstrap(app: App, span: Span) -> None: app.get("/", home, name="home") app.mount("/static", StaticAssets(Path("./static"), name="static"))After registration, app.url_for("home") is "/" and app.url_for("static:js/app.js") resolves to the fingerprinted URL (the name="static" on the mount defines the static:… reverse keys).
class App(*, middleware=())
Concrete app type: everything on Router plus errors, reverse URLs, and shutdown-aware tasks.
Uncaught exceptions become HTTP responses only before headers are sent; after that, telemetry still records the failure. Use create_task for work tied to a running server so graceful shutdown can observe it.
async App.__call__(c, w)
Protocol entrypoint: open a span, resolve routes, handle errors, always call w.end().
Parameters
- c—Request context (
app,req,span,route,state). - w—Response writer for this message on the connection.
Notes
Trailing slashes (except /) get 308 to the router-normalized path (same rules as find_handler); wrong method on a matching path yields 405. If a registered error handler raises, the failure is logged and a 500 is sent when no response has started yet.
App.on_error(exc_type, handler)
Register a handler for uncaught exceptions of type exc_type (subclasses use MRO; most specific wins).
Parameters
- exc_type—Exception class to match.
- handler—Receives
(context, writer, exc); may returnNoneor an awaitable.
Notes
Only runs while the writer has not started (w.started is false); HttpException is registered by default.
App.create_task(coro, *, name=None)
Schedule a coroutine on the running loop and retain the task until it completes.
Parameters
- coro—Coroutine to run.
- name—Optional task name for debuggers.
Returns
The new asyncio.Task.
Raises
StarioError: If no event loop is running (call from async request or app code only).
App.url_for(name, *, params=None, query=None)
Build a path (and optional query string) from a registered name.
Parameters
- name—Value passed as
name=toget/handleor from mountedStaticAssets. - params—
str.formatmapping for{placeholders}in the stored pattern. - query—Values merged with
urllib.parse.urlencode(lists/tuples repeat the key).
Returns
A path starting with /, or host/... style when the route used host matching, plus ?... when query is set.
Raises
StarioError: If name is unknown or placeholders are missing from params.
async App.join_tasks()
Await until every task created with create_task has finished (including nested scheduling).
Useful in tests to wait for background work after the HTTP response has been sent. Do not call from inside a coroutine that is itself tracked in create_task or you risk deadlock.
Server
Server is the asyncio listener around an App instance: it builds the app, runs normalize_bootstrap on bootstrap(app, span), binds TCP or Unix, handles signals, serves until SIGINT/SIGTERM, then shuts down with a shared grace window for connections and app.create_task work, forced close, cancellation of remaining tasks, and bootstrap teardown.
Under the hood
run() creates the app once, opens a startup span, enters bootstrap and _server_scope, calls start_serving, then blocks until shutdown. Shutdown records open connections, force-closed transports, and cancelled tasks on the span. The span passed to bootstrap is the same object for the whole run; when shutdown begins, a linked shutdown span is created and the span id is repointed so late telemetry still attributes correctly (see Application).
If event_loop_name is set (for example via stario --loop), it appears as server.event_loop on startup attributes.
Bootstrap contract
normalize_bootstrap() accepts the return shapes documented under Application lifecycle above and wraps them in one async with for Server, tests, and other callers.
normalize_bootstrap(bootstrap)
Wrap a user bootstrap(app, span) so it always acts as an async context manager.
Parameters
- bootstrap—Callable returning an async context manager, a single-yield async generator, an awaitable, or
None.
Returns
AppBootstrap — use as async with wrapped(app, span): from Server or tests.
Raises
StarioError: From the wrapper when the return type is unsupported or an async generator mis-yields.
Notes
Async generators must yield exactly once between setup and teardown.
class Server(bootstrap, tracer, *, app_factory=None, host='127.0.0.1', port=8000, graceful_timeout=5.0, backlog=2048, unix_socket=None, unix_socket_mode=432, compression=CompressionConfig(min_size=512, zstd_level=3, zstd_window_log=None, brotli_level=4, brotli_window_log=None, gzip_level=6, gzip_window_bits=None), event_loop_name=None, max_request_header_bytes=65536, max_request_body_bytes=10485760)
Binds a listener, runs the bootstrap context around a fresh app, serves until SIGINT/SIGTERM, then drains.
async Server.run()
Block until SIGINT/SIGTERM (or fatal error): create app, enter bootstrap, serve, drain, tear down.
Raises
StarioError: If run is called twice on the same instance, or bootstrap suppresses startup errors.
Notes
Requires an already-running event loop. Signals are temporarily replaced for SIGINT/SIGTERM during the call.
Static assets
StaticAssets scans a directory once at construction (usually from bootstrap with app.mount("/static", StaticAssets(...))). That pass fingerprints files, builds metadata, optionally loads small files into RAM, and registers a catch-all route under the prefix. There is no per-request directory walk.
Fingerprinting
Each file is hashed with xxHash64; the public name is {stem}.{digest}{suffix} with relative paths preserved. Requests for logical paths get a 307 to the fingerprinted URL. Immutable responses use strong cache headers by default. That redirect is what browsers see when they hit the logical path; app.url_for("static:…") emits the fingerprinted path directly so HTML does not pay an extra round trip—same behavior as reverse URLs for named routes.
Reverse URLs for static files
Pass name="static" (or another prefix) so each file has a logical key such as static:js/app.js. app.url_for("static:js/app.js") emits the fingerprinted URL in HTML without an extra redirect (Application). Without name=, files are still served but url_for has no entry.
Compression at startup
Eligible files may be pre-compressed at wire-up. Brotli and zstd are preferred for quality on modern clients; gzip is also generated with configurable levels (defaults in code: zstd 9, brotli 9, gzip 7, with window settings tuned for static assets). Already-compressed types (images, fonts, .woff2, and similar) skip recompression. Large files stay on disk and stream without holding full precompressed blobs.
Mounting
StaticAssets is re-exported from the stario package (same as App, Context, etc.).
from pathlib import Path from stario import App, Span, StaticAssets async def bootstrap(app: App, span: Span) -> None: app.mount("/static", StaticAssets(Path("./static"), name="static"))class StaticAssets(directory='./static', *, cache_control='public, max-age=31536000, immutable', cache_max_size=1048576, name=None, hash_chunk_size=4194304, compress_min_size=256, filesystem_chunk_size=65536, zstd_level=9, zstd_window_log=21, brotli_level=9, brotli_window_log=22, gzip_level=7, gzip_window_bits=15)
Scan a directory at init: fingerprint filenames, cache small files (optional pre-compression), stream large files.
Mount with app.mount("/static", static); exposes root / named_routes like a router subtree. Non-fingerprint paths 307 to hashed URLs.
async StaticAssets.__call__(c, w)
GET/HEAD handler: resolve {path...} against the startup index, redirect, 404, or send bytes from memory or disk.