Application¶
The application object and its configuration.
Bases: AsgiMixin, DispatchMixin, ErrorsMixin, IntrospectionMixin, LifecycleMixin, MCPMixin, MiddlewareMixin, MountingMixin, OpenAPIMixin, PluginsMixin, ServingMixin, TestingMixin, BackgroundTasksMixin, TemplatingMixin, Router
Ultra-fast async web framework.
Usage::
app = Veloce()
@app.get("/")
async def index(request: Request):
return {"message": "Hello, World!"}
app.run()
property
¶
The app's shared Jinja2 Environment.
Available once a template_folder has been configured (either
via the constructor or by binding Jinja2Templates). Mutate it
directly to register filters/globals or tweak settings:
app.jinja_env.filters["money"] = fmt. Raises RuntimeError
when no templating is configured.
property
¶
The app's Jinja template loader.
The FileSystemLoader (or whatever loader the bound
Jinja2Templates env uses). None when no templating is
configured - Veloce returns None for an app with no template
folder rather than raising.
property
¶
middlewares: tuple[Middleware, ...]
The registered Middleware instances, in the order they will run.
Registration order unless priorities were set, in which case this is the priority order the pipeline actually uses. Checking whether a middleware is installed - a plugin guarding against registering itself twice, or a startup check that CORS is present - otherwise means reading a private list.
Standard ASGI middleware classes are not here; they are wrapped around the app when the ASGI stack is assembled rather than run per request by this pipeline.
Usage::
if not any(isinstance(m, CORSMiddleware) for m in app.middlewares):
app.add_middleware(CORSMiddleware, allow_origins=["*"])
property
¶
A {endpoint_name: handler} view of registered routes.
Endpoint names follow a simple rule - the route's name=
kwarg, or the handler's __name__ when no name is set; blueprint
routes are prefixed with <bpname>.. Returned dict is a fresh
snapshot - mutation doesn't poison framework state.
property
¶
Inspection view of registered error handlers.
Returns a {blueprint_name_or_None: {key: handler}} mapping.
App-level handlers live under the None key; each blueprint's
handlers live under the blueprint's name, keyed by integer status
code or exception class. Blueprint handlers are scoped to their own
routes at dispatch time, so they appear under their blueprint name
here, not folded into None.
property
¶
View of registered before_request hooks.
Returns {blueprint_name_or_None: [hook, ...]}. App-level hooks
live under the None key; blueprint hooks under the blueprint's
name. The dispatcher walks the None bucket plus the bucket
whose name matches the matched route's endpoint prefix.
property
¶
Return the per-blueprint after-request hook registry.
property
¶
Return the per-blueprint teardown-request hook registry.
property
¶
Snapshot mapping of bp.name -> Blueprint.
Returns a fresh copy, so caller mutations don't affect the framework. Re-registering the same name overwrites the previous entry.
property
¶
View of registered URL-value preprocessors.
Returns {blueprint_name_or_None: [fn, ...]} - app-level processors
under None, then each blueprint's under its dotted name. A nested
blueprint's entry is the flattened chain that applies to its routes,
outermost first, which is what runs.
property
¶
View of registered URL-default callbacks, keyed as url_value_preprocessors.
property
writable
¶
Whether debug mode is enabled; bound to config['DEBUG'].
Interprets a dotenv-style string (DEBUG=false) correctly rather than
treating any non-empty string as truthy.
property
writable
¶
Session-signing secret; bound to config['SECRET_KEY'].
SessionMiddleware constructed without an explicit secret_key=
resolves it from here on the first request, so app.secret_key = ...
and config['SECRET_KEY'] are one and the same setting.
property
¶
url_map: URLMap
Read-only mapping of registered URL rules.
Iterating it yields URLRule objects (rule, methods, endpoint).
Subscript by endpoint name (app.url_map["users.detail"]) returns
a list of rules registered under that endpoint. Length is the
total registered route count.
This is the introspection-friendly view of Veloce.routes;
callers who just want the dict-list keep using app.routes.
property
¶
The registered instrumentation hooks, in the order they will run.
The read half of add_instrumentation, which had none: a caller
checking whether it has already installed its own hook - the
idempotency guard veloce.otel performs - otherwise has to read the
private list the dispatch core iterates.
A tuple, so a caller cannot register or reorder through it; use
add_instrumentation for that.
property
writable
¶
Active JSONProvider instance.
Lazily instantiated from app.json_provider_class so swapping
encoders is just: app.json_provider_class = MyJSONProvider.
Setting app.json = instance replaces it explicitly.
The provider serialises every value a handler returns - a dict, a
list, a model, a msgspec struct, a (body, status) tuple, jsonify,
and a JSONResponse subclass named by response_class - so one dialect
covers the application. An app that configures nothing keeps the direct
orjson path and pays nothing for the indirection.
It does not reach the framework's own wire formats: cache keys always sort so equal mappings hash alike, and signed cookies, JWTs and protocol frames are not the application's to restyle.
property
¶
Filesystem path of the directory containing import_name's module.
Named package_root rather than root_path because Veloce.root_path
already means the ASGI mount prefix, which is a different thing entirely
- one is a location on disk, the other a URL prefix. Useful for resolving
template and static directories relative to the app's source file.
property
¶
Writable instance folder beside the package.
Veloce resolves <package_root>/instance as a per-deployment
writable directory for config, SQLite files, uploads, etc.
An explicit instance_path= constructor argument overrides
this computed default. The directory is not auto-created -
the caller decides whether to mkdir it.
property
¶
Accessor that returns the veloce.signals module.
Veloce ships its signals as module-level singletons, so this
attribute returns the module - app.signal_namespace.request_started
is the same Signal instance as veloce.signals.request_started.
property
writable
¶
Callable that raises typed HTTPExceptions by status code.
app.aborter(404) is equivalent to the module-level abort(404) helper.
It is a distinct attribute so applications can subclass Aborter to add
custom code-to-exception mappings: assign to app.aborter, or mutate the
instance this returns.
One instance per application, built on first access and kept - so a mutation to its mapping sticks, and does not reach any other app.
property
¶
True after the first request has been fully handled.
Read-only compatibility accessor. Useful when conditional setup
depends on whether the app has bootstrapped yet, e.g. a
before_first_request hook firing exactly once is reflected
here as True.
property
¶
Click Group for app-defined custom CLI commands.
Accessing app.cli lazily constructs a click.Group once.
Custom commands attach via the standard Click decorator:
@app.cli.command("init-db")
def init_db():
...
The veloce console script automatically discovers and mounts
the group as a custom subcommand when launched with an app
reference. click is required at access time but not at import
time - the ImportError is deferred and produces a useful
message instead of a hard-import crash on environments that
don't need the CLI.
property
writable
¶
Mutable map of dependency callables to test replacements.
Populate it to swap a real dependency for a fake one in tests::
app.dependency_overrides[get_db] = get_fake_db
The resolver consults this map on every request, so changes take
effect immediately. Assigning a fresh dict (or calling .clear())
removes all overrides.
route(path: Annotated[str, Doc('URL path template, including `{param}` / `{param:converter}` placeholders.')], methods: Annotated[list[str] | None, Doc('HTTP methods this handler serves; defaults to `GET`.')] = None, dependencies: Annotated[list[Any] | None, Doc('Dependencies run for this route, appended after the router-level ones.')] = None, response_model: Annotated[Any, _DOC_RESPONSE_MODEL] = _INFER_RESPONSE_MODEL, tags: Annotated[list[str] | None, Doc('OpenAPI tags for this route, combined with the router-level tags.')] = None, summary: Annotated[str | None, Doc('Short OpenAPI summary for this operation.')] = None, name: Annotated[str | None, Doc("Endpoint name for `url_for` reverse lookup; defaults to the handler's name.")] = None, description: Annotated[str | None, Doc("OpenAPI description; defaults to the handler's docstring.")] = None, deprecated: Annotated[bool, Doc('Mark the operation as deprecated in the OpenAPI document.')] = False, response_description: Annotated[str, Doc('Description of the successful response in the OpenAPI document.')] = MSG_SUCCESSFUL_RESPONSE, status_code: Annotated[int, Doc('Default HTTP status code for a successful response.')] = HTTP_200_OK, response_class: Annotated[Any, Doc('Response class for this route, overriding the router and framework defaults.')] = None, response_model_include: Annotated[set[str] | None, Doc('Fields to include when serializing the response model.')] = None, response_model_exclude: Annotated[set[str] | None, Doc('Fields to exclude when serializing the response model.')] = None, response_model_exclude_unset: Annotated[bool, Doc('Omit fields left unset on the response model from the serialized output.')] = False, response_model_exclude_defaults: Annotated[bool, Doc('Omit fields equal to their default on the response model from the serialized output.')] = False, response_model_by_alias: Annotated[bool, Doc('Serialize the response model using field aliases instead of attribute names.')] = False, response_model_exclude_none: Annotated[bool, Doc('Omit fields whose value is `None` from the serialized response model.')] = False, include_in_schema: Annotated[bool, Doc('Register the route but omit it from the generated OpenAPI document when False.')] = True, responses: Annotated[dict[int, dict[str, Any]] | None, Doc('Additional OpenAPI responses for this route, overlaid on the router-level ones.')] = None, operation_id: Annotated[str | None, Doc('Explicit OpenAPI `operationId`; defaults to the route name.')] = None, openapi_extra: Annotated[dict[str, Any] | None, Doc("Arbitrary dict deep-merged into this route's OpenAPI operation object.")] = None, defaults: Annotated[dict[str, Any] | None, Doc('Fixed values merged into the path params at dispatch without overriding URL-matched ones.')] = None, callbacks: Annotated[dict[str, Any] | None, Doc("OpenAPI Callback objects emitted verbatim into the operation's `callbacks` field.")] = None, strict_slashes: Annotated[bool | None, Doc('When False, match both slashed and unslashed forms; `None` defers to the app policy.')] = None, subdomain: Annotated[str | None, Doc('Constrain the route to a subdomain of `SERVER_NAME`; `*` matches any non-apex subdomain.')] = None, host: Annotated[str | None, Doc('Constrain the route to an exact `Host` header value (case-insensitive).')] = None, expose_as_mcp_tool: Annotated[bool, Doc('Expose the route as an MCP tool in the contrib MCP registry.')] = False, mcp_description: Annotated[str | None, Doc("LLM-facing description for the route's MCP tool, required when exposed as one.")] = None, expose_as_mcp_resource: Annotated[bool, Doc('Expose the read-only route as an MCP resource in the contrib MCP registry.')] = False, mcp_resource_uri: Annotated[str | None, _DOC_MCP_RESOURCE_URI] = None, mcp_resource_mime_type: Annotated[str | None, _DOC_MCP_RESOURCE_MIME_TYPE] = None, mcp_meta: Annotated[dict[str, Any] | None, _DOC_MCP_META] = None, mcp_resource_size: Annotated[int | None, Doc("Size in bytes advertised for the route's MCP resource.")] = None, mcp_resource_annotations: Annotated[dict[str, Any] | None, Doc("Annotations (audience, priority) advertised for the route's MCP resource.")] = None, mcp_scopes: Annotated[Sequence[str] | None, Doc('Authorization scopes required to call this route over MCP.')] = None, mcp_icons: Annotated[Sequence[Any] | None, Doc('Optional MCP `Icon` objects a client may render next to the tool/resource.')] = None, mcp_task_support: Annotated[bool, _DOC_MCP_TASK_SUPPORT] = False, exclude_middleware: Annotated[Sequence[str | type] | None, _DOC_EXCLUDE_MIDDLEWARE] = None, stream: Annotated[bool, _DOC_STREAM] = False) -> Callable
Register a route for any set of HTTP methods.
exclude_middleware=["CSRFMiddleware"] opts this route out of the
named middleware (matched against each middleware's name), so a
webhook or health-check route can skip CSRF, auth, or rate limiting
without forking the middleware. Routes that declare no exclusions
pay no extra per-request cost.
GET route decorator. Safe and idempotent - RFC 9110 Sec. 9.3.1.
PUT route decorator. Idempotent - RFC 9110 Sec. 9.3.4.
DELETE route decorator. Idempotent - RFC 9110 Sec. 9.3.5.
HEAD route decorator. Like GET with no body - RFC 9110 Sec. 9.3.2.
OPTIONS route decorator - RFC 9110 Sec. 9.3.7.
QUERY route decorator - RFC 10008.
QUERY is safe and idempotent like GET but carries a request body like
POST, for read-only operations whose parameters do not fit a URL (search,
filtering, paging). The handler reads the body exactly as a POST handler
does (request.get_json() / a body model parameter).
websocket(path: Annotated[str, Doc('URL path template for the WebSocket route, including `{param}` placeholders.')]) -> Callable
Register a WebSocket route via decorator.
websocket_listener(path: str, *, receive: str = 'json', send: str = 'json', on_connect: RouteHandler | Callable[..., Any] | None = None, on_disconnect: RouteHandler | Callable[..., Any] | None = None) -> Callable
Register a WebSocket route wrapping a per-message callback.
The decorated callback handles one message at a time; the framework
owns the accept handshake, the receive loop, and the clean close on
disconnect. The callback is called as cb(data), or cb(ws, data)
when its first parameter is named ws/socket (or it takes two
positional parameters). Returning a non-None value sends it back in
send mode; returning None sends nothing.
receive/send select the codec ("json" default, or "text" /
"bytes"). on_connect(ws) runs after accept; on_disconnect(ws)
always runs when the loop ends, including on peer disconnect. Sync
callbacks and hooks are offloaded to the executor.
Usage::
@app.websocket_listener("/echo")
async def echo(data):
return data
For full control over the handshake and loop use @app.websocket.
add_websocket_route(path: Annotated[str, Doc('URL path template for the WebSocket route, including `{param}` placeholders.')], handler: Annotated[RouteHandler, Doc('Callable invoked with the accepted WebSocket connection when the route matches.')]) -> None
Register a WebSocket route imperatively (ASGI shape).
The non-decorator form of @app.websocket(path).
Register an imperative WebSocket route, mirroring add_api_route.
The non-decorator form of @app.websocket(path). name, when given,
registers the route for reverse lookup so app.url_for(name) resolves
to its path.
add_api_route(path: str, endpoint: RouteHandler, *, methods: list[str] | None = None, **kwargs: Any) -> None
Register a route imperatively.
The non-decorator form: the handler argument is named endpoint
here and forwarded to add_route (where it is handler). All
route kwargs - response_model, tags, dependencies,
status_code, openapi_extra, ... - pass straight through.
Defaults to ["GET"] when methods is omitted.
Build the URL for name, applying the @app.url_defaults callbacks.
They run before delegating to Router.url_for, so injected defaults
appear in the rendered URL.
On build failure (unknown endpoint or missing path parameter),
each registered app.url_build_error_handlers callback is
invoked with (error, endpoint, values) in order; the first
non-None return is used. If none recovers, a BuildError is
raised.
Resolve a URL path by endpoint name and parameters.
iter_routes(*, include_hidden: bool = False) -> list[tuple[str, str, RouteInfo]]
Return every registered route as (method, path, info).
app.routes is a summary view: six fields of RouteInfo's full
record, which is enough to render a route table and not enough for
anything that inspects a route. This returns the records themselves, so
response models, dependencies, security requirements and the rest are
reachable without touching private state.
Hidden routes - WebSocket routes and those registered
include_in_schema=False - are omitted unless include_hidden is
set; the default is the schema-visible set.
Usage::
for method, path, info in app.iter_routes():
if info.response_model is not None:
print(method, path, info.response_model)
async
¶
Handle one request - run the middleware chain, then route dispatch.
cp is the compiled pipeline for this request. __call__ already
resolves it (to gate the ASGI wrapper stack) and threads it in so the
generation check runs once per request, not once here and once there.
A caller that reaches this method directly (a mounted sub-app, the
public dispatch_request aliases) passes None and the pipeline is
resolved here.
log_exception(exc: BaseException, request: Request | None = None) -> None
Log an exception with traceback.
Routes the exception through the app logger at ERROR level. Used internally before falling back to a 500 response; exposed publicly so error-handler code can re-log via the same path.
request names the request that failed, which is most of the value of
the record: a traceback with no path is hard to place in a live log.
Callers with no request in hand (a background task, a CLI hook) omit it.
Silencing this is logging.getLogger(app.import_name).setLevel(...) or
any other ordinary logging configuration - it is the app's own logger,
deliberately, so an operator turns it down the way they turn down
anything else.
make_default_options_response(path: str, allowed_methods: list[str] | None = None) -> Response
Build the auto-OPTIONS response for path.
Returns a 200 response with an empty body and an Allow header
listing every method registered for path, augmented with
HEAD (whenever GET is supported) and OPTIONS itself per
RFC 9110 Sec. 9.3.7. Callers that register an explicit OPTIONS
handler can use this to compose the default Allow set. Pass
allowed_methods when the registered set is already known to skip
the redundant get_allowed_methods lookup.
Return an async context manager driving the lifespan cycle.
async with app.lifespan_context(): ... runs the full startup
sequence (lifespan CM enter + on_startup handlers) on entry
and the shutdown sequence on exit - independent of any request.
Useful for tests and for embedding the app where you want
startup/shutdown without an ASGI server in the loop.
spawn(coro: Annotated[Coroutine[Any, Any, Any], Doc('Coroutine to run as a background task.')], *, name: Annotated[str | None, Doc('Name the task is registered under; anonymous when omitted.')] = None) -> Task[Any]
Schedule a long-lived, app-scoped background task.
Unlike per-request background tasks, a spawned task lives for the
application's lifetime: it is tracked with a strong reference (so the
loop cannot GC it mid-flight) and is cancelled-and-drained during
shutdown, honouring the GRACEFUL_TASK_TIMEOUT config budget per
task. Pass name to make the task retrievable and cancellable by
name via get_spawned_task / cancel_spawned_task; a duplicate name
raises. Failures are logged through the same path as request-scoped
background tasks, so app and request work surface uniformly.
Must be called with a running event loop (e.g. from within an
on_startup handler, the lifespan CM, or a request); calling it
before the loop exists raises RuntimeError.
Usage::
@app.on_startup
async def _start_poller():
app.spawn(poll_queue(), name="queue-poller")
Register a template context processor.
The function should return a dict that merges into the template context.
Register a function as a Jinja filter.
Usage::
@app.template_filter("upper")
def upper(s): return s.upper()
The filter becomes available in every Jinja2Templates render that
runs inside this app's request scope. name defaults to the
function's own __name__.
Register a callable as a Jinja global, reachable by name in a template.
Same shape as template_filter.
Imperative equivalent of @template_global.
Register a Jinja test - used in {% if x is name %} constructs.
Imperative equivalent of @template_filter.
Imperative equivalent of @template_test.
Merge registered context-processor output into context.
Runs every @app.context_processor callback and folds the
returned dicts into context in place, without overriding
keys the caller already set (the documented semantics - explicit context
wins). Returns the same dict for chaining.
Return the named spawned task, or None if there is no such task.
Cancel a named spawned task. Return whether a task was cancelled.
supervise(coro_factory: Annotated[Callable[[], Coroutine[Any, Any, Any]], Doc('Zero-argument callable returning a fresh coroutine on every restart.')], *, name: Annotated[str, Doc('Name the supervised task is registered under.')], max_restarts: Annotated[int, Doc('Restarts allowed inside `restart_window` before giving up.')] = 5, restart_window: Annotated[float, Doc('Seconds the restart budget is counted over.')] = 60.0, backoff: Annotated[float, Doc('Seconds to wait before the first restart.')] = 1.0, max_backoff: Annotated[float, Doc('Ceiling the doubling backoff is clamped to.')] = 30.0) -> Task[Any]
Run a long-lived coroutine, restarting it on failure.
coro_factory is a zero-argument callable that returns a fresh
coroutine each time it is invoked - the supervisor calls it to start
the task and again to restart after a crash, so a single coroutine
object (which cannot be re-awaited) is not accepted. The supervised
coroutine is expected to run for the application's lifetime; if it
returns normally the supervisor restarts it, and if it raises the
failure is logged and the coroutine is restarted after a bounded
backoff delay. asyncio.CancelledError is never suppressed, so the
task stops cleanly when cancelled at shutdown.
A count-within-window circuit breaker bounds runaway restarts: at most
max_restarts restarts are allowed within any restart_window seconds.
The restart counter resets whenever the coroutine runs for longer than
the window without failing (a clean run), so steady-state restarts far
apart never trip the breaker; a tight crash loop does. When the breaker
trips the supervisor logs the give-up and stops restarting. backoff
is the initial delay between restarts and doubles up to max_backoff
on consecutive failures, resetting to backoff after a clean run.
The supervisor itself runs as an app.spawn(...) task, so it is tracked
with a strong reference and cancelled-and-drained on shutdown like any
other spawned task. name is required (the supervisor task is named so
it is retrievable / cancellable via get_spawned_task /
cancel_spawned_task); a duplicate name raises. Must be called with a
running event loop.
Usage::
@app.on_startup
async def _start():
app.supervise(lambda: poll_queue(), name="queue-poller")
async
¶
Wait for every currently-spawned background task to finish.
Returns True when they all completed, False when timeout elapsed
first. Tasks are left running either way - this waits, it does not
cancel; _drain_spawned_tasks is the shutdown path that does.
A response's background task runs after the response is sent, which is the point of it, so a caller that needs its effect - a test asserting the email was queued, a script that must not exit early - otherwise has to guess at a sleep or drive the loop by hand. Newly spawned tasks are picked up too: a task that spawns another is waited for in full.
Usage::
await app.wait_for_background_tasks()
test_client(**kwargs: Any) -> TestClient
Return an in-memory TestClient for this app.
app.test_client() is the factory API; the kwargs (e.g.
follow_redirects=True, base_url=...) are forwarded to
TestClient.__init__. Equivalent to TestClient(app, **kwargs)
for callers that prefer the method form.
async_test_client(**kwargs: Any) -> AsyncTestClient
Return an AsyncTestClient for this app.
The async counterpart of test_client() - used as
async with app.async_test_client() as client: inside an async
test, so requests are awaited on the test's own running loop
rather than driven through a private loop. Kwargs are forwarded
to AsyncTestClient.__init__.
Bind current_app and reset g for use outside a request.
Use as with app.app_context(): .... CLI commands, background
jobs, and tests need this when they want to read app.config or
write into g without going through handle_request. Nestable:
the previous binding (if any) is restored on exit.
test_request_context(path: str = '/', method: str = HTTP_METHOD_GET, headers: dict[str, str] | None = None, query_string: str = '', body: bytes = b'') -> _TestRequestContext
Synthesise a fake request for outside-request testing.
Inside with app.test_request_context(): ..., current_app, g,
and the request-scoped contextvars resolve as if Veloce
had just received that request - without spinning up the full
dispatch pipeline. Strict subset of what handle_request does:
no middleware, no DI, no handler.
run(host: Annotated[str | None, Doc('Interface to bind; defaults to `127.0.0.1`. Conflicts with `bind_all`.')] = None, port: Annotated[int, Doc('TCP port to listen on.')] = 8000, workers: Annotated[int, Doc('Must be `1`; the built-in server runs a single process.')] = 1, access_log: Annotated[bool, Doc('Print the start-up banner and install the development access log.')] = True, ssl_context: Annotated[SSLContext | None, Doc('Serve HTTPS for local testing when given.')] = None, bind_all: Annotated[bool, Doc('Bind every interface (`0.0.0.0`) instead of localhost.')] = False, reload: Annotated[bool, Doc('Restart the server whenever a project `.py` file changes.')] = False) -> None
Start the built-in development server.
Veloce's from-scratch HTTP server is intended for local
development only. For production, run the app under a hardened
ASGI server - uvicorn your_module:app - which veloce is fully
compatible with through its ASGI __call__ interface.
run() logs a reminder of this on startup.
host resolves to "127.0.0.1" when unset so the dev server
is reachable only from the local machine. Pass bind_all=True
to opt in to all-interfaces binding ("0.0.0.0"). host and
bind_all=True are mutually exclusive - passing both raises
ValueError to avoid silent privilege widening. Binding to
0.0.0.0 exposes the dev server to every reachable network -
including remote attackers if the machine is on a public network
- so it should be used only in trusted environments and never
with debug=True.
ssl_context - an ssl.SSLContext - turns on HTTPS for local
testing; it is handed straight to loop.create_server(ssl=...).
Left None (the default) the serving path is byte-for-byte the
same as plain HTTP. Production should still terminate TLS at
uvicorn or a reverse proxy.
workers must be 1: the built-in server runs a single process
and does not pre-fork. Passing more raises ValueError - run under
uvicorn module:app --workers N or the gunicorn VeloceWorker for
multiple processes.
reload=True turns on the development auto-reloader: this process
supervises a child that serves requests and restarts it whenever a
project .py file changes. The watching happens in the supervisor, so
the served child carries no overhead. It is a development aid - leave it
off for any deployment.
Install plugin and return it.
Call plugin.install(self). When plugin has a truthy name,
record it as self.extensions[name] and raise ValueError if that
name is already taken. Raise TypeError when plugin has no callable
install. A named plugin is recorded only after its install returns,
so a failed install leaves no partial registry entry.
Return the generated OpenAPI schema dict.
Computes the schema on first call, caches the result in
app.openapi_schema. Subsequent calls return the cached dict;
users can mutate the result in place (e.g. to inject custom
info.x-logo or tags orderings) and the swagger UI / json
endpoints will serve the mutated copy.
To bypass the auto-build entirely, assign a custom dict to
app.openapi_schema before any request lands.
mount(prefix: Annotated[str, Doc('Full path the sub-application is served at.')], app: Annotated[Any, Doc('A `Veloce` sub-app, an ASGI application, or a `StaticFiles`.')], *, expose_mcp: Annotated[bool, Doc("Also publish the sub-app's MCP tools through this app's server.")] = False) -> None
Mount a sub-application at a path prefix.
prefix is the full path the sub-app is reached at. The app's own
Veloce(prefix=...) does not apply here - that prepends to routes this
app registers, and a mount places another application rather than
registering a route. An app built with Veloce(prefix="/api") and
mount("/sub", child) serves the child at /sub, not /api/sub; write
mount("/api/sub", child) for that.
A veloce sub-app is dispatched through the parent's request
pipeline. Any other ASGI application - an ASGI micro-app, an
instrumentation shim - is dispatched at the ASGI layer instead:
the matched prefix is stripped from the scope's path and moved
onto root_path, so the mounted app sees a normal root-relative
request.
Lifecycle: a mounted Veloce sub-app has its startup and shutdown
driven by the parent - the parent runs each child's startup after its
own during lifespan/run() startup, and tears children down in
reverse on shutdown, so a child's on_startup / lifespan resources
initialise and release without a separate ASGI lifespan. A mounted
non-Veloce ASGI app receives http and websocket scopes only:
the parent does not fan the lifespan cycle out to it, so it must
not depend on ASGI lifespan events for its setup. A mounted ASGI
app owns its entire prefix subtree - a native route registered under
the same prefix is unreachable.
Prefixes must not overlap: registering a prefix equal to, nested
under, or containing an existing mount raises ValueError, since
overlapping mounts would shadow each other in a confusing,
order-dependent way.
expose_mcp=True additionally publishes the sub-app's MCP tools,
resources and prompts through the parent's MCP server, with tool and
prompt names prefixed by the mount. It is opt-in because mounting an app
for its HTTP routes should not silently hand an agent everything it can
do.
mount_static(prefix: Annotated[str, Doc('URL path the directory is served under.')] = '/static', directory: Annotated[str, Doc('Directory to serve, absolute or relative to the package root.')] = 'static', html: Annotated[bool, Doc('Serve `index.html` for a directory request, in HTML mode.')] = False, must_exist: Annotated[bool, Doc('Refuse a missing directory; `False` downgrades it to a warning.')] = True) -> None
Mount a static file directory.
The directory must exist and be readable at wiring time (a typo
otherwise 404s every asset silently); pass must_exist=False to
downgrade the check to a warning when the directory is created after
the app is constructed.
add_middleware(middleware: Annotated[Any, Doc('A `Middleware` instance, a middleware class, or an ASGI wrapper class.')], **options: Annotated[Any, Doc('Keyword arguments the middleware class is built with.')]) -> None
Add middleware to the pipeline.
Call forms:
add_middleware(VeloceMiddlewareClass, **options)- a class subclassingMiddlewareis instantiated with**optionsand appended to the request/response pipeline.add_middleware(instance)- append an already-builtMiddlewareinstance directly.add_middleware(ASGIMiddlewareClass, **options)- a class that is not aMiddlewaresubclass is treated as a standard ASGI middleware: it wraps the whole application and is instantiated asASGIMiddlewareClass(app, **options)when the ASGI stack is assembled. This is what lets third-party ASGI middleware (observability, tracing, profiling, ...) plug in. Middleware added first is the outermost wrapper.
Pass name= to override the instance's exclusion name (the identifier
exclude_middleware=[...] on a route references). The override is
applied after construction rather than forwarded into the subclass
constructor, so per-instance naming works for every Middleware
subclass - including user subclasses whose __init__ does not accept a
name keyword.
Pass priority= (an int, default 0) to order this middleware
deterministically regardless of registration order. Higher priority
runs earlier in the request phase and correspondingly later in the
response phase; middleware of equal priority keeps registration order
(a stable tiebreak). The ordered chain is resolved once at registration
time, so per-request dispatch pays no sorting cost. When no middleware
sets a priority the behaviour is unchanged - the chain is the plain
registration order it has always been. priority applies to the
request/response Middleware pipeline only, not to ASGI-class
middleware (which is ordered by its own wrap nesting).
Register a middleware on the (request, call_next) -> response chain.
Accepts a BaseHTTPMiddleware instance, a bare callable, or a class
(which is instantiated with no args). Returns the registered object so
it can be used as a decorator.
middleware(middleware_class_or_type: type | str, **kwargs: Any) -> Callable[[Callable], Callable] | None
Add middleware - supports both a class form and a decorator form.
The two forms return different things: the decorator form returns the
decorator, and the class form returns None because it is a
statement, not a decorator. Writing @app.middleware(CORSMiddleware)
therefore fails at decoration rather than silently doing nothing.
Class form: app.middleware(CORSMiddleware, allow_origins=["*"]) Decorator form: @app.middleware("http") async def add_header(request, call_next): response = await call_next(request) response.headers["X-Custom"] = "value" return response
mcp_tool(description: Annotated[str, Doc('Required LLM-facing description, separate from the docstring.')], *, name: Annotated[str | None, Doc("Tool name; defaults to the decorated function's own.")] = None, namespace: Annotated[str | None, Doc('Prefix qualifying the tool name, for a shared catalogue.')] = None, scopes: Annotated[Sequence[str] | None, Doc('Scopes the calling principal must hold to see and call it.')] = None, tags: Annotated[Sequence[str] | None, Doc('Free-form labels published with the tool.')] = None, icons: Annotated[Sequence[Icon] | None, Doc('Icons a client may render beside the tool.')] = None, task_support: Annotated[bool, Doc('Allow the call to be started as a long-running MCP task.')] = False, annotations: Annotated[dict[str, Any] | None, Doc('MCP tool annotations, such as `readOnlyHint`.')] = None, meta: Annotated[dict[str, Any] | None, Doc('Opaque `_meta` published with the tool.')] = None, version: Annotated[str | None, Doc('Version string published with the tool.')] = None) -> Callable[..., Any]
Register an MCP-only tool callable by an AI agent (contrib.mcp).
The decorated coroutine (or sync function) becomes an MCP tool whose
input JSON Schema is derived from its signature; Depends() params
resolve through the same dependency machinery routes use, with an
MCPContext standing in for the HTTP Request. description is the
required LLM-facing text (separate from the docstring). namespace
prefixes the tool name (<namespace>_<name>), mirroring how a
blueprint namespaces an exposed route. icons is an optional list of
Icon objects a client may render next to the tool. task_support=True
lets a client run the tool as a background task (task-augmented
tools/call, polled via tasks/get / tasks/result). version labels
this registration: two tools sharing a name and declaring different
versions are both registered, the higher one is listed, and a call
naming no version reaches it.
Usage::
@app.mcp_tool(description="Add two integers")
async def add(a: int, b: int) -> int:
return a + b
mcp_prompt(description: str, *, name: str | None = None, namespace: str | None = None, scopes: Sequence[str] | None = None, icons: Sequence[Icon] | None = None, meta: dict[str, Any] | None = None) -> Callable[..., Any]
Register an MCP prompt template fetchable by an AI agent (contrib.mcp).
The decorated callable's parameters become the prompt's arguments, and its
return - a string, or a list of role/content messages - becomes the
messages prompts/get returns. Depends() params resolve through the
same dependency machinery routes use, with an MCPContext standing in for
the HTTP Request. description is the required LLM-facing text;
namespace prefixes the prompt name (<namespace>_<name>). icons is an
optional list of Icon objects a client may render next to the prompt.
Usage::
@app.mcp_prompt(description="Summarise a topic in three bullets")
async def summarise(topic: str) -> str:
return f"Summarise {topic} in three bullet points."
Register an already-built MCPTool (contrib.mcp).
The decorator builds a tool from a handler; this takes one that already
exists - most often from derive_tool, which narrows a registered tool
into the façade an agent should see::
app.add_mcp_tool(derive_tool(internal, name="search", arguments={...}))
Register a hook that runs before every MCP call (contrib.mcp).
Called with the primitive's name and the arguments it was given. Return
None to let the call proceed, or any other value to answer with that
instead of invoking the handler - the same short-circuit shape
before_request has. Raising an MCPError reports the failure to the
client, which is how an authorization check refuses a call.
Unlike before_request, this reaches a tool registered with
@app.mcp_tool, which has no route and so no request lifecycle::
@app.before_mcp_call
async def audit(name, arguments):
log.info("mcp call", extra={"tool": name})
Register a hook that runs after every MCP call (contrib.mcp).
Called with the primitive's name and the handler's return value, and returns the value to send on - so a hook may rewrite a result, or return it unchanged. Hooks run in registration order, each seeing what the last returned. It does not run when the call raised.
mcp_completer(*, argument: str, prompt: str | None = None, resource: str | None = None) -> Callable[..., Any]
Register an argument-value completer for an MCP prompt or resource (contrib.mcp).
The decorated callable suggests values for one argument of a prompt
(named) or a resource (by URI template) as the user types, answering the
MCP completion/complete request. It is called with the partial value
and a mapping of the sibling argument values already resolved, and returns
a sequence of candidate strings (or a CompletionResult for explicit
totals). Pass exactly one of prompt or resource. An argument with no
registered completer answers with an empty completion.
Usage::
@app.mcp_completer(prompt="greet", argument="name")
async def complete_name(value: str, context: dict[str, str]) -> list[str]:
return [n for n in KNOWN_NAMES if n.startswith(value)]
mount_mcp(transport: Annotated[str, Doc('One of `stdio`, `http`, or the deprecated `sse`.')] = 'stdio', *, path: Annotated[str, Doc('Where an HTTP or SSE transport mounts; ignored on stdio.')] = '/mcp', auth: Annotated[Any, Doc('An `MCPAuth` making an HTTP endpoint an OAuth 2.1 resource server.')] = None, principal: Annotated[Any, Doc('`Principal` whose identity and scopes the served tools run under.')] = None, allowed_origins: Annotated[Sequence[str] | None, Doc('Enable `Origin` validation, the DNS-rebinding defense.')] = None, exclude_middleware: Annotated[Sequence[str] | None, Doc('App middleware the transport routes opt out of.')] = None, sessions: Annotated[bool, Doc('Opt into the `Mcp-Session-Id` lifecycle.')] = False, resumable: Annotated[bool, Doc('Opt into SSE resumability, replaying missed events on reconnect.')] = False, tool_filter: Annotated[Any, Doc('`(tool, principal) -> bool` narrowing what `tools/list` reports.')] = None, cache_ttl_ms: Annotated[int | None, Doc('Freshness hint sent with cacheable results; `0` marks them stale.')] = None, page_size: Annotated[int | None, Doc('Opt the list methods into cursor pagination at this page size.')] = None, tool_search: Annotated[bool, Doc('Publish `search_tools` / `describe_tools` / `run_tools` in place of the catalogue.')] = False, session_backend: Annotated[Any, Doc('Store sharing HTTP sessions between workers, with async `read`/`write`/`delete`.')] = None, message_path: Annotated[str, Doc('URL an SSE stream names for the client to POST to; ignored elsewhere.')] = '/messages') -> Any
Build the MCP server and serve the registered tools.
Assembles the tool registry from @app.mcp_tool registrations plus every
route flagged expose_as_mcp_tool=True, the resource registry from every
read-only route flagged expose_as_mcp_resource=True, and the prompt
registry from @app.mcp_prompt registrations, then serves them over the
chosen transport. Call this after the tool / resource / prompt routes are
registered.
Transports
transport="stdio" (the default) serves JSON-RPC 2.0 on stdin/stdout for
subprocess use and returns an awaitable serve coroutine that runs until
stdin closes, inside the app's lifespan_context() - so every
on_startup handler runs before the first tool is served. Schedule it
explicitly (asyncio.run(app.mount_mcp())).
transport="http" mounts the Streamable HTTP transport as a POST route
at path (default /mcp) on this app and returns None; serve the app
with any ASGI server (or app.run()) as usual.
transport="sse" mounts the deprecated split-endpoint wire of MCP
revision 2024-11-05, for a client that speaks only that: a GET at path
(defaulting to /sse) opens a stream that names message_path as the URL
to POST to, each POST is acknowledged 202 and its JSON-RPC response
arrives on the stream. Prefer transport="http" for anything new - one
endpoint, and a dropped connection can be resumed.
Arguments, in signature order:
path is where an HTTP or SSE transport mounts; ignored on stdio.
auth (a veloce.contrib.mcp.MCPAuth) makes an HTTP endpoint an
OAuth 2.1 resource server - validating the bearer token on every request
and serving the RFC 9728 metadata.
principal (a veloce.Principal) establishes the identity and scopes the
served tools run under. A local subprocess is trusted, so this is how a
stdio server takes its identity from the environment.
allowed_origins enables Origin validation - the DNS-rebinding defense.
exclude_middleware names app middleware the transport routes opt out of
(an app-wide auth middleware that auth replaces, for instance).
sessions opts into the Mcp-Session-Id lifecycle: the server assigns a
session id on initialize, requires it on later requests (400 missing,
404 once terminated), and accepts a DELETE to terminate it.
resumable opts into SSE resumability: each streamed event gets an id
encoding its stream, and a GET carrying Last-Event-ID replays only
that stream's missed events, so a client can reconnect after a dropped
connection.
tool_filter narrows what tools/list reports per caller beyond the
declared scopes: a callable (tool, principal) -> bool (sync or async)
that hides tools an agent has no business seeing, so its context is not
spent on tools it cannot invoke. Declared scopes are applied first,
whether or not a filter is set - every list omits what this caller would
be refused - so a filter can only hide further, never reveal. Hiding a
primitive does not change what happens if it is called anyway.
cache_ttl_ms sets the freshness hint sent with cacheable results
(tools/list, prompts/list, resources/list, resources/read and
server/discover) on the modern protocol revision; 0 marks them
immediately stale. A list that can differ between callers is additionally
marked private, so a shared proxy cannot serve one caller's answer to
another.
page_size opts the list methods into cursor pagination: each answers
with at most that many entries plus a nextCursor while more remain, so
a large catalogue reaches the agent a page at a time instead of filling
its context in one response. Left unset, every list is answered in full -
a client may ignore nextCursor, so paginating uninvited would hide the
rest of the catalogue from one that does.
tool_search publishes three tools in place of the catalogue -
search_tools, describe_tools and run_tools - so a server with a
large catalogue spends the agent's context on the tools it turns out to
need rather than on every tool it has. run_tools executes declared
calls, not code: each step names a registered tool and its arguments, and
a step's argument may reference an earlier step's result.
session_backend shares HTTP sessions between workers - any object with
async read / write / delete methods over a SessionRecord. Without
one a session lives in the worker that minted it, so a request reaching a
different worker is answered 404 and the client starts a new session.
message_path is the URL an SSE stream names for the client to POST to;
ignored on the other transports.
Register a function to run before each request.
Register a function to run exactly once on the first request.
A legacy hook style - lifespan startup handlers are preferred,
but first-request hooks are still a common pattern,
so both are supported. Hooks fire serially in registration
order; single-fire is guarded with an asyncio.Lock so
concurrent first requests don't double-run the callbacks.
Register a function to run after each request.
Register a function to run after request teardown.
Called with an optional exception argument, even if an exception occurred.
Register a function to run on app-context teardown.
Register startup/shutdown event handlers.
Deprecated: use @app.on_startup / @app.on_shutdown instead.
Scheduled for removal in v1.0.0.
Register an additional lifespan context manager.
lifespan= is a single slot owned by the application, which leaves a
plugin or blueprint no way to own a resource with paired setup and
teardown - it has to split the pair across on_startup / on_shutdown
and lose the try/finally (and the yielded handle) between them.
factory is called with the app and must return an async context
manager. Every registered lifespan is entered on the same exit stack as
lifespan=, so teardown runs in reverse registration order, a failure
part-way through startup unwinds only what was entered, and teardown
errors are aggregated rather than masking one another.
The yielded value is not consumed - a plugin holds its own handle, the same way it holds any other state it owns.
Usage::
class BrokerPlugin:
name = "broker"
def install(self, app):
app.add_lifespan(self.lifespan)
@contextlib.asynccontextmanager
async def lifespan(self, app):
broker = await connect()
try:
yield {"broker": broker}
finally:
await broker.close()
Imperative event-handler registration - ASGI shape.
Deprecated: call app.on_startup(fn) / app.on_shutdown(fn)
directly instead. Scheduled for removal in v1.0.0.
Register a coroutine to run once at app startup. Alias of on_startup.
Register a coroutine to run once at app shutdown. Alias of on_shutdown.
Attach a function as the view for an already-registered name.
Useful when separating route declaration (via
app.add_url_rule(rule, endpoint="x")) from view registration.
Replaces the existing route's handler in place.
Iterate over every registered Blueprint.
Returns the blueprints in registration order (Python 3.7+ dict insertion order). Yields the Blueprint objects, not their names.
Register a function returning a dict to merge into veloce shell.
each processor is called with no args; its dict
becomes part of the namespace the interactive shell starts with.
Useful for surfacing models / db sessions / common helpers so
User.query.first() works without a manual from myapp.models
import User every time.
Build the dict the CLI's shell command drops into.
Always includes app (this Veloce instance) and g. Each
registered shell-context-processor's return dict overlays on
top, in registration order - later processors win on conflicts.
Register a callback that may mutate the matched path params.
Called as fn(endpoint, values) before the handler runs.
Usage::
@app.url_value_preprocessor
def pull_lang(endpoint, values):
from veloce import g
g.lang = values.pop("lang", "en")
endpoint is the route name; values is the path_params dict
(mutating it in place is the supported way to remove / rewrite
values before the handler sees them).
Register a callback injecting default kwargs into every URL build.
Called as fn(endpoint, values) from url_for and url_path_for.
Usage::
@app.url_defaults
def add_lang(endpoint, values):
from veloce import g
values.setdefault("lang", g.get("lang", "en"))
Runs in registration order; mutate values in place.
Register an error handler without a decorator.
The key is an int status code or an exception class. Anything else is
refused: a non-class key landed in _exception_handlers, which is matched
by walking a raised exception's MRO, so it could never be found.
exception_handlers={"404": h} - realistic when the mapping is read from
JSON, TOML or the environment - registered without a word and never fired.
Register a custom exception handler by exception type or status code.
Imperative exception-handler registration - ASGI shape.
The non-decorator form of @app.exception_handler(...).
Accepts an exception class (matched by MRO at dispatch time) or
an int HTTP status code.
async
¶
handle_http_exception(exc: HTTPException, request: Request | None = None) -> Response
Build the response for an HTTPException.
Walks registered status-code + class handlers first (matching
abort() semantics), falling back to JSON
{"detail": exc.detail, "status_code": exc.status_code} with
exc.headers applied - byte-identical to what the request cycle
emits for the same exception, so a handler reached over MCP or from
a background task reports the error exactly as it does over HTTP.
Pass request= when calling from inside a request scope so the
registered error handler receives the real failing request
(with the actual path, method, path_params, state, etc.)
instead of a synthetic GET /. Callers without a request (the
original out-of-band use case) can omit it.
async
¶
Dispatch an arbitrary exception.
HTTPException -> handle_http_exception. Otherwise walks
registered class handlers (MRO); on no match, logs via
log_exception and returns 500. Pass request= to propagate
the real failing request to the registered handler; omit to
get a synthetic GET / for out-of-band callers (background
tasks, CLI hooks).
async
¶
full_dispatch_request(request: Request) -> Any
Dispatch request through the full before/after-request hook chain.
An alias for _dispatch_request, which already runs the chain inline.
async
¶
preprocess_request(request: Request) -> Any
Run all before_request hooks for request.
Walks the registered hooks in order; if any hook returns a
non-None value it short-circuits the chain and that value is
returned (the contract - a non-None return becomes the
response). Both sync and async hooks are supported. App-level
hooks fire first, then the matched-blueprint bucket - the
same shape _dispatch_request uses.
async
¶
Run all after_request hooks for (request, response).
Hooks fire in reverse registration order; each hook may return a
replacement Response (the contract: any other return keeps the
existing one). App-level hooks reverse-iterate first, then the matched
blueprint's, then the request's one-shot after_this_request callbacks.
This is the dispatch path itself, not a re-implementation of it, so a hook behaves here exactly as it will in production - including the signature adaptation that lets a hook declare only the arguments it wants.
staticmethod
¶
Wrap func so it is callable from synchronous code.
- If
funcis a regular function, returns it unchanged. - If
funcis a coroutine function, returns a sync wrapper that runs the coroutine to completion on a dedicated event loop and returns the result.
Use to bridge async handlers / hooks into sync code (CLI commands, background workers, test scaffolding).
make_response(value: Any) -> Response
Coerce a handler-return value into a Response.
Accepts (with this coercion table):
- Response -> returned as-is
- str / bytes -> wrapped as a text/HTML response
- dict / list -> wrapped as a JSON response via jsonify
- tuple of (body, status), (body, status, headers) or
(body, headers) -> unpacked and re-coerced
- anything else -> JSON, matching what dispatch does with the same
value returned from a handler
A tuple of any other length is not a response tuple and is answered as
a plain value. veloce.make_response and dispatch read the same table
(_unpack_response_tuple); dispatch keeps its own fast lanes for the
shapes a handler returns most, but answers alike.
Register a route. See Router.add_route for the full signature.
This override exists only to bracket the base implementation: refuse a
mutation once the app is serving, and drop the route caches the new
route invalidates. It forwards everything untouched, so the parameters,
defaults and Doc(...) annotations are the base method's.
__doc__ and __signature__ are re-pointed at the base just below, so
help(app.add_route) and an editor's signature hint show the documented
surface rather than (*args, **kwargs). Restating the parameter list
here instead would add a ninth hand-maintained copy of it.
include_router(router: Annotated[Router, Doc('Sub-router to mount; a `Blueprint` brings its hooks and handlers with it.')], prefix: Annotated[str, Doc('Path prefix every mounted route is registered under.')] = '', url_prefix: Annotated[str | None, Doc('Blueprint-style spelling of `prefix`; wins when both are given.')] = None) -> None
Mount a sub-router under an optional path prefix.
Accepts either a Blueprint (delegates to register_blueprint,
honouring its hooks / error handlers / url processors) or a plain
Router (delegates to Router.include_router).
prefix and url_prefix are interchangeable and name the same thing;
both spellings are accepted so router-style and blueprint-style calling
code can each use the one that reads naturally, and url_prefix wins
when both are given.
add_instrumentation(hook: Callable | None = None, *, exclude_routes: Iterable[str] | None = None) -> Callable
Register an observability instrumentation hook.
hook is called once per finished HTTP request with a
RequestMetrics record - the request method, the concrete path,
the matched route template (a low-cardinality metric label), the
status code, and the wall-clock duration in milliseconds. It may be
a plain function or a coroutine function. A hook that raises is
logged and skipped, so instrumentation never breaks a response.
Returns hook unchanged, so it also works as a decorator. Both the
no-argument and the keyword-argument decorator forms are supported -
when hook is omitted a decorator is returned that captures
exclude_routes and registers the function it wraps:
@app.add_instrumentation
def export(metrics):
statsd.timing(metrics.route or "unmatched", metrics.duration_ms)
@app.add_instrumentation(exclude_routes={"/health"})
def export(metrics):
statsd.timing(metrics.route or "unmatched", metrics.duration_ms)
Pass exclude_routes to suppress this hook for noisy routes - a set
of matched route templates (e.g. {"/health", "/metrics"}). When a
finished request's route template is in the set the hook is skipped,
so health checks and scrape endpoints never pollute traces or metric
series. Matching is on the low-cardinality template resolved during
routing (never the concrete, attacker-controlled path), so there is
no per-request regex and no path-normalisation bypass. The filter is
applied in the core delivery loop, so every consumer of this hook -
tracing, metrics, access logs, custom - honours the same exclusion.
An unmatched request (route template None) is never excluded by a
named-route set.
With no hook registered the request path carries no instrumentation cost - not even a clock read.
Register component to report to veloce check and security_audit.
Middleware reports on itself through Middleware.audit without being
registered anywhere - it is already in the stack. Something that hardens
or exposes the app without being middleware has nowhere to say so, and
a mounted MCP endpoint is the case in tree: it registers routes, so
without this the audit had nothing to ask about a tool-execution
endpoint with no authentication.
component needs an audit(ctx) yielding Findings, exactly as a
middleware's does; it may also carry the Auditable class markers
(sets_hardening_headers, audit_needs_routes). Returns component,
so it can be used as a decorator on a class.
Usage::
class TelemetryEndpoint:
def audit(self, ctx):
if not self.authenticated:
yield Finding(
"telemetry endpoint accepts unauthenticated writes",
severity="error",
fix="pass auth=...",
id="telemetry-unauthenticated",
)
app.register_auditable(TelemetryEndpoint())
Return human-readable warnings about the current security posture.
A rendering of veloce.audit.run(self), which is the structured form -
severity, remedy and a stable id per finding. Use that where a tool
needs to tell an error from a warning; this stays a list of lines
to print. An empty list means nothing was flagged.
Middleware reports on itself through Middleware.audit, so a
middleware written outside this package is audited on the same terms
as a built-in one. What stays invisible is hardening the app does not
own - a reverse proxy terminating TLS, or a middleware that reports
nothing - so a clean audit is a statement about this app's middleware,
not about the deployment around it.
Return human-readable findings about each route's response contract.
A rendering of the response-model-contradiction and
routes-undocumented findings from veloce.audit.run(self), which is
the structured form. A route whose response_model= names a different
model than its return annotation is a warning; a route with no
response contract at all is info, because many such routes are
legitimate - HTML pages, redirects, streams.
An empty list means nothing was flagged.
Serve a file from app.static_folder.
app.static_folder defaults to "static" (relative to
app.package_root). Use app.static_url_path to control the
URL prefix when mounting via app.static(...). Returns a
FileResponse; traversal-safe via safe_join.
This reads the file synchronously and emits a
DeprecationWarning when called on a running loop. From async
handlers, prefer send_static_file_async.
async
¶
Serve a file from app.static_folder - async variant.
Reads the file in an executor via send_from_directory_async, so
it never blocks the event loop. Prefer this from async handlers
over the sync send_static_file.
Return a Click CliRunner bound for testing app.cli.
Veloce exposes this for unit-testing @app.cli.command(...)
handlers without manual Click import. Kwargs flow through to
click.testing.CliRunner.
register_blueprint(blueprint: Annotated[Blueprint, Doc('Blueprint whose routes and hooks are mounted.')], url_prefix: Annotated[str | None, Doc("Path prefix to mount under; defaults to the blueprint's own.")] = None) -> None
Mount a Blueprint's routes + hooks onto this app.
- Re-registers each route under
(url_prefix or bp.url_prefix) + pathso the same blueprint can be mounted twice (e.g. v1/v2 versions). - Buckets the blueprint's
before_request/after_request/teardown_requesthooks under its name, so they fire only for that blueprint's own routes. They are kept apart from the app-level lists rather than spliced into them and filtered: a per-request scan of every hook is what bucketing avoids. - Buckets blueprint-level error handlers under the blueprint name (and each nested child under its dotted name), scoped to that blueprint's own routes; an app-level handler still catches everything as a fallback.
Mountable multiple times on different apps with different prefixes - the blueprint itself stays unmodified.
add_url_rule(rule: Annotated[str, Doc('URL path template, including `{param}` / `{param:converter}` placeholders.')], endpoint: Annotated[str | None, Doc('Endpoint name for `url_for`; required when registering an endpoint-only stub.')] = None, view_func: Annotated[Callable | None, Doc('Handler for the route; `None` registers an endpoint-only stub for later attachment.')] = None, methods: Annotated[list[str] | None, Doc('HTTP methods this rule serves; defaults to `GET`.')] = None, **kwargs: Any) -> None
Add a URL rule programmatically.
view_func=None registers an endpoint-only stub: the route
exists for url_for resolution but has no handler yet. Attach
one later with @app.endpoint(endpoint). Calling such a route
before a handler is attached raises a clear RuntimeError.
Requires endpoint to be set in the stub case.
Return the dependency override mapping.
Bases: dict[str, Any]
A dict that knows how to load itself from common config sources.
Only keys made of ASCII uppercase letters, digits, or underscores
(and not starting with a digit) are stored - see _is_uppercase_key.
staticmethod
¶
Return the documented default config keys with their values.
Seeded into app.config at construction so reads never raise
KeyError. Values are the documented defaults; veloce-specific
behaviour reads several of these (MAX_CONTENT_LENGTH,
JSON_SORT_KEYS, PROPAGATE_EXCEPTIONS).
Bulk-update from mapping and/or kwargs.
Only UPPERCASE keys are config keys. A non-uppercase keyword argument
raises: it was typed out one key at a time, so dropping it silently means
from_mapping(debug=True) leaves DEBUG untouched and says nothing. Keys
in a mapping are filtered quietly instead, because a settings dict or a
parsed config section legitimately carries entries that are not config.
Always returns True so the call can be used as a chaining sentinel.
Import UPPERCASE attributes from a module, class, instance, or dotted-path string.
from_object("myapp.settings.Prod") resolves the dotted path,
then walks attributes whose names pass _is_uppercase_key.
Execute a Python file and pull UPPERCASE module-level names.
Returns True on success. If silent=True and the file is
missing, returns False instead of raising.
Load KEY=VALUE pairs from a dotenv-style .env file.
Full-line # comments and blank lines are skipped, an optional
export prefix is accepted, and a value wrapped in matching
single or double quotes is unquoted. An unquoted value may carry
a trailing # inline comment, which is stripped; a # inside
quotes is kept literal. A .env file carries no types, so a value for
a key with a known type is converted to it: DEBUG=false stores False
rather than a truthy string, and MAX_CONTENT_LENGTH=1000 stores 1000
rather than "1000". An unparseable number raises, naming the key. Only
UPPERCASE keys are kept (see from_mapping). With silent=True a missing file returns
False rather than raising.
Keys are stored exactly as the file spells them, and os.environ is
not touched. This does not compose with from_prefixed_env, which
strips its prefix: a file setting MYAPP_TIMEOUT becomes the config key
MYAPP_TIMEOUT here and TIMEOUT there. veloce run seeds
os.environ from the same file before importing the app, so an app
using both reads two different keys depending on how it was started -
pick one of the two and use it on every path.
Read a filename from os.environ[varname] and from_pyfile it.
Load config from the env vars named <prefix>_..., less the prefix.
Values are JSON-decoded, falling back to the raw string when JSON
parsing fails. Nested config uses the __ separator:
VELOCE_MAIL__SERVER sets config["MAIL"]["SERVER"].
Reads os.environ only. from_env_file reads a file and keeps each key
verbatim, so the two name the same setting differently - see its note.
A value that is not valid JSON is given the type its key is read as, the same way the file loader does; a value that cannot be converted raises, naming the key. Nested keys have no declared type and are stored as read.
from_file(filename: str, load: Callable[[Any], Mapping[str, Any]] = _orjson_load, silent: bool = False, text: bool = False) -> bool
Load any structured file (JSON, TOML via tomllib.load, YAML ...).
Opens the file in text or binary mode (per text=), hands the
file object to load, expects a mapping back, then applies it
through from_mapping.
get_namespace(namespace: str, *, lowercase: bool = True, trim_namespace: bool = True) -> dict[str, Any]
Return all config keys starting with namespace, trimmed.
A helper for extracting one subsystem's settings.
With lowercase=True (default), trimmed keys are lower-cased
- extension code conventionally uses lowercase attribute names.
Bases: Protocol
A Veloce plugin: any object exposing install(self, app).
Usage::
class TimingPlugin:
name = "timing"
def install(self, app):
app.add_instrumentation(self._record)
app.install(TimingPlugin())
Serve liveness and readiness probes, and gate readiness on shutdown.
Usage::
from veloce import Veloce
from veloce.health import HealthPlugin
app = Veloce()
health = app.install(HealthPlugin())
@health.readiness_check("cache")
async def cache_ready() -> bool:
return await redis.ping()
/livez reports whether the process and its event loop are running; it
deliberately ignores dependency checks, because restarting a container
cannot fix someone else's database.
/readyz reports whether this replica should receive traffic: startup has
completed, shutdown has not begun, and every registered check passes. A
failing check yields 503 with a per-check body naming what failed, so a
probe failure is diagnosable from the response alone rather than only from
logs.
Checks run concurrently and share one timeout; a check that hangs is
reported as failed rather than holding the probe open until the
orchestrator's own timeout fires.
readiness_check(name: Annotated[str, Doc('Name reported for this check in the probe body.')]) -> Callable[[ReadinessCheck], ReadinessCheck]
Register a readiness check under name.
The check returns True when this replica can serve traffic. Raising is treated as not-ready, so a check does not need its own try/except.
Mark the replica as draining so /readyz starts failing.
Call this when a shutdown signal arrives, before connections are drained: the orchestrator then stops routing new requests here while in-flight ones finish. Veloce's own shutdown calls it automatically.
Signals¶
The pub/sub primitives and the eight signals Veloce fires around the request
and app-context lifecycle. Connect a receiver with
request_started.connect(fn); see the
Signals guide for the payload each one carries.
A named pub/sub signal - standard shape.
Receivers connect via connect(receiver, sender=ANY_SENDER) and
detach via disconnect(receiver, sender=ANY_SENDER).
send(sender, **kwargs) fires every receiver subscribed for that
exact sender (compared by is, falling back to ==) plus every
receiver subscribed for ANY_SENDER. Return values are collected
into a list of (receiver, value) tuples so callers can introspect
what fired, though veloce's own code ignores the return value.
asend and send_robust_async await async receivers concurrently
(sync receivers still run inline in registration order first).
Register receiver to fire when send(sender) runs.
sender=ANY_SENDER (the default) subscribes to every send.
Pass a specific sender to filter - the receiver then only fires
when send is called with that exact sender. Returns the
receiver unchanged so it can be used as a decorator.
Remove the subscription for (receiver, sender).
Mirrors connect - to detach a per-sender subscription pass the
same sender. With the default sender=ANY_SENDER it removes
any subscription matching receiver, regardless of which sender
it was bound to (back-compat with the previous unfiltered API).
Targeted detach matches the stored sender directly, not via
_matches - _matches is the send-time rule ("does this
subscription fire for that sender?"), where a stored
ANY_SENDER deliberately matches every send. Reusing that rule
in disconnect would silently delete an ANY_SENDER
subscription whenever the caller targeted a specific sender.
send(sender: Any = None, **kwargs: Any) -> SignalResult
Fire receivers subscribed for sender (and for ANY_SENDER).
Returns (receiver, value) pairs in registration order. With no
subscriptions the call short-circuits, so callers can invoke
send unconditionally rather than guarding with
has_receivers_for - a single live-scan then both fires and
prunes dead weakrefs.
send_robust(sender: Any = None, **kwargs: Any) -> SignalResult
Like send, but never aborts on a failing receiver.
Returns (receiver, value) pairs in registration order. The
second tuple element is the receiver's return value, OR an
Exception instance if the receiver raised. Per-receiver
exceptions are logged at WARNING and substituted into the
result list so the caller can inspect failures while subsequent
receivers still fire.
Sync-only: if a receiver is an async function (or otherwise
returns a coroutine), the coroutine is closed and a TypeError
is recorded in the result list instead. Use send_robust_async
to await async receivers.
async
¶
asend(sender: Any = None, **kwargs: Any) -> SignalResult
Async, non-robust send - awaits async receivers concurrently.
Returns (receiver, value) pairs in registration order. Sync
receivers run inline immediately, preserving registration order
and raising on the first sync error exactly like send. Async
receivers are collected and awaited concurrently, each inside a
copy of the dispatch-time context. Like send, the first failing
receiver propagates its exception (non-robust contract); use
send_robust_async to capture per-receiver failures instead.
Even on the non-robust path every async receiver runs to
completion before this coroutine returns OR raises: the concurrent
run collects all results (failures included), and only afterwards
is the first exception, in receiver order, re-raised. This
guarantees no receiver is still touching request-scoped state once
asend has returned - a return_exceptions=False gather would
instead re-raise the first failure while later receivers kept
running in the background past teardown.
async
¶
send_robust_async(sender: Any = None, **kwargs: Any) -> SignalResult
Async variant of send_robust - awaits async receivers concurrently.
Returns (receiver, value) pairs in registration order. The
second tuple element is the receiver's return value, OR an
Exception instance if the receiver raised. Sync receivers run
inline first, in registration order, each wrapped so a raised
exception is recorded as its result. Async receivers (or any
receiver returning a coroutine) are then awaited concurrently;
one failing receiver never cancels the others. Per-receiver
exceptions, raised either at call time or while awaiting, are
logged at WARNING and substituted into the result list.
How many receivers this signal holds, live or not yet pruned.
The counting counterpart to has_receivers_for, which answers the
live question and short-circuits. This counts what is stored, so a
weak reference whose target has been collected is still counted until
send prunes it - which is the distinction anyone debugging a leak, or
checking that pruning happened, actually needs.
True if any connected receiver would fire for sender.
A side-effect-free predicate that short-circuits on the first
live, matching receiver. Dead weakrefs are skipped but not pruned
here; pruning is left to send / _iter_live_targets.
A factory that returns named Signal instances, one per name.
Calling signal(name) repeatedly with the same name returns the
same Signal object, so independent parts of an application can
obtain a shared signal by agreeing on a name rather than passing the
instance around.
Usage::
from veloce.signals import Namespace
signals = Namespace()
user_registered = signals.signal("user-registered")
@user_registered.connect
def welcome(sender, **kw):
...
user_registered.send(app, user=user)
signal(name: str, doc: str | None = None) -> Signal
Return the Signal named name, creating it on first use.
Repeated calls with the same name return the identical instance, and
the first call's doc is the one kept - a later call naming a different
one does not rewrite a signal other code already holds.
doc is recorded on the signal as Signal.doc, not accepted and
discarded - which would make it a silent no-op for anyone porting code
that passes it.