Application¶
The application object and its configuration.
Bases: AsgiMixin, DispatchMixin, ErrorsMixin, LifecycleMixin, 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
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
¶
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
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 jsonify(...) and anything else that asks it
for bytes. A handler returning a bare dict or list takes the direct
orjson path instead and does not consult it, so a custom dialect - key
sorting, a house encoder - does not reach those responses. Return
jsonify(...) from a handler whose dialect must apply.
property
¶
Filesystem path of the directory containing import_name's module.
Veloce exposes this as app.root_path; veloce already uses
Veloce.root_path for the ASGI mount prefix, so we surface the
package-directory variant under a non-conflicting name. Useful
for resolving template / static directories relative to the
app's source file.
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
¶
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; veloce returns a fresh Aborter instance per access
so users can mutate _mapping per-app without affecting others.
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
¶
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, ...]}. Veloce flattens
blueprint preprocessors into the app list at registration time,
so the dict carries a single None key.
property
¶
View of registered URL-default callbacks.
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("Type used to filter and serialize the handler's return value and the OpenAPI response schema. Defaults to the handler's return annotation when it names a model; pass `None` to declare no response contract.")] = _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("Resource URI for the route's MCP resource: a static URI, or a URI template (`users://{user_id}`) binding its path parameters.")] = None, mcp_resource_mime_type: Annotated[str | None, Doc("Media type advertised for the route's MCP resource.")] = None, mcp_meta: Annotated[dict[str, Any] | None, Doc("`_meta` published on this route's MCP tool or resource.")] = 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("Allow this route's MCP tool to run as a background task (task-augmented `tools/call`, polled via `tasks/get` / `tasks/result`).")] = False, exclude_middleware: Annotated[Sequence[str] | None, Doc('Names of middleware this route opts out of.')] = None, stream: Annotated[bool, Doc('Opt into request-body streaming: the body is not buffered before the handler, so the handler may consume `request.stream()` incrementally.')] = False) -> Callable
Generic route decorator.
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.
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
Declarative WebSocket route - wrap 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.
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 - accessible from any
template by name. 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.
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: Callable[[], Coroutine[Any, Any, Any]], *, name: str, max_restarts: int = 5, restart_window: float = 60.0, backoff: float = 1.0, max_backoff: float = 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")
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.
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: str | None = None, port: int = 8000, workers: int = 1, access_log: bool = True, ssl_context: SSLContext | None = None, bind_all: bool = False, reload: bool = 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 a sub-application at a path prefix.
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: str = '/static', directory: str = 'static', html: bool = False, must_exist: bool = 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 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 BaseHTTPMiddleware-style middleware on the
(request, call_next) -> response chain. Accepts an 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.
Add middleware - supports both a class form and a decorator form.
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
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.
Register a coroutine to run once at app shutdown.
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.
Register an error handler without a decorator.
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.
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.
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).
Mount a sub-router include_router.
Accepts either a Blueprint (delegates to register_blueprint,
honouring its hooks / error handlers / url processors) or a
plain Router (delegates to Router.include_router). The
prefix and url_prefix are interchangeable; both spellings
spells it prefix, Veloce spells it url_prefix.
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.
Apply a security-hardened configuration baseline.
- Marks the session cookie
Secure,HttpOnly, and (unless already configured)SameSite=Lax. - Registers
SecurityHeadersMiddleware-nosniff, frame-deny, a referrer policy, and a one-year HSTS max-age - unless one is already present.
Call once after construction, before serving. Production-oriented:
the Secure cookie flag means cookies are not sent over plain
HTTP, so do not call this for local HTTP development.
Return human-readable warnings about the current security posture.
An empty list means nothing was flagged. Drives the
veloce check CLI command and is also callable directly from a
pre-deploy script or a test.
Report routes whose declared response contract is absent or contradictory.
A response contract is only checkable once every route is registered, so it is reported here rather than discovered by a request: a handler whose return value cannot satisfy its declared model would otherwise surface as a server error in production, one request at a time.
Two findings are produced. A route whose explicit response_model=
names a different model than its return annotation is a contradiction -
the annotation tells a reader and a type checker one thing while the wire
carries another. A route with no response contract at all is listed so an
app cannot silently ship most of its surface undocumented; many such
routes are legitimate (HTML pages, redirects, streams), so this is
informational rather than a failure.
An empty list means nothing was flagged. Drives veloce check and is
callable directly from a pre-deploy script or a test.
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.
async
¶
full_dispatch_request(request: Request) -> Any
Alias for _dispatch_request (which already runs the
full before/after-request hook 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
¶
process_response(request: Request, response: Any) -> Any
Run all after_request hooks for (request, response).
Hooks fire in reverse registration order; each hook may
return a replacement response (the contract: a None return
keeps the existing response). App-level hooks reverse-iterate
first, then the matched-blueprint bucket - mirrors
_dispatch_request's ordering.
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,), (body, status), (body, status, headers),
or (body, headers) -> unpacked and re-coerced
Decorator attaching a function as the view for name
on an already-registered route.
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 function fn(endpoint, values) that can mutate the
matched path params 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).
Veloce.url_for runs @app.url_defaults callbacks 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.
Register a function fn(endpoint, values) that injects default
kwargs into every url_for / url_path_for call.
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.
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). - Splices the blueprint's
before_request/after_request/teardown_requesthooks into the app's own lists. Blueprint hooks fire only for blueprint-routed requests (gated viarequest.endpointstarting with"<bpname>."); we tag the blueprint's hooks so the dispatcher can filter. - 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.
mcp_tool(description: str, *, name: str | None = None, namespace: str | None = None, scopes: Sequence[str] | None = None, tags: Sequence[str] | None = None, icons: Sequence[Icon] | None = None, task_support: bool = False, annotations: dict[str, Any] | None = None, meta: dict[str, Any] | None = None, version: str | None = None) -> Callable
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
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.
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: str = 'stdio', *, path: str = '/mcp', auth: Any = None, principal: Any = None, allowed_origins: Sequence[str] | None = None, exclude_middleware: Sequence[str] | None = None, sessions: bool = False, resumable: bool = False, tool_filter: Any = None, cache_ttl_ms: int | None = None, page_size: int | None = None, tool_search: bool = False, session_backend: Any = None, message_path: str = '/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.
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())). A local subprocess is trusted,
so authentication is from the environment: pass a principal (a
veloce.Principal) to establish the identity / scopes the served tools run
under.
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. Pass auth (a
veloce.contrib.mcp.MCPAuth) to make the endpoint an OAuth 2.1 resource
server - validating the bearer token on every request and serving the
RFC 9728 metadata. allowed_origins enables Origin validation
(DNS-rebinding defense); exclude_middleware names app middleware the
transport routes opt out of (an app-wide auth middleware auth replaces).
sessions opts into 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.
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.
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.
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.
Call this after the tool / resource / prompt routes are registered.
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
¶
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 stored; lowercase keys are silently skipped. 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. Values are stored as plain strings -
a .env file carries no types. 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.
Pull env vars starting with <prefix>_, strip the prefix, store
with JSON-decoded values (falling back to the raw string when JSON
parsing fails). Nested config via __ 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.
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.
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.
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
¶
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
¶
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.
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)