faust.contrib.fastapi

Compatibility alias for faust.contrib.asgi. New code should import the framework-neutral module directly.

Compatibility imports for the renamed ASGI integration.

The integration does not depend on FastAPI specifically. New code should import from faust.contrib.asgi; this module remains as an alias for the older, framework-specific import path.

exception faust.contrib.fastapi.LoopMismatch

The Faust app is bound to an event loop other than the running one.

class faust.contrib.fastapi.AsgiService(asgi_app: Any = None, *, host: Optional[str] = None, port: Optional[int] = None, uvicorn_options: Optional[Mapping[str, Any]] = None, **kwargs: Any)

Serve an ASGI application with uvicorn, on the Faust worker’s loop.

Usually created for you by serve_asgi().

faust_app: Optional[AppT] = None

Faust app supplying late-bound worker web settings.

driver_version: str = 'ASGI'

Driver description used in the worker banner.

server_shutdown_timeout: float = 10.0

Seconds to wait for the server to finish serving on shutdown.

opentelemetry: Optional[bool] = None

Instrument the ASGI app with OpenTelemetry. None auto-detects.

asgi_app: Any = None

The ASGI application to serve.

host: Optional[str] = None

Interface to bind to.

port: Optional[int] = None

Port to bind to.

uvicorn_options: Mapping[str, Any] = {}

Extra keyword arguments for uvicorn.Config.

classmethod get_web_url() → URL

Return the configured URL without instantiating the server.

Return type:

URL

logger: logging.Logger = <Logger faust.contrib.asgi (WARNING)>

log: CompositeLogger

diag: DiagT

async_exit_stack: AsyncExitStack

exit_stack: ExitStack

async on_start() → None

Start serving.

Return type:

None

async on_stop() → None

Ask the server to exit and wait for it.

Return type:

None

property label: str

Return description of this service, used in logs. :rtype: str

class faust.contrib.fastapi.FaustLifespanMiddleware(asgi_app: Any, faust_app: AppT, **kwargs: Any)

Add Faust startup and shutdown to an ASGI application.

This is the framework-neutral alternative to faust_lifespan() for applications that do not expose a lifespan-constructor hook. Django is the common example:

from django.core.asgi import get_asgi_application
from faust.contrib.asgi import FaustLifespanMiddleware

django_app = get_asgi_application()
application = FaustLifespanMiddleware(django_app, faust_app)

HTTP and WebSocket scopes are passed to asgi_app unchanged. The middleware owns the ASGI lifespan scope, starting Faust before reporting startup complete and stopping it before reporting shutdown complete.

Use faust_lifespan() instead when the inner framework already has a lifespan that must also run. This middleware intentionally does not pass lifespan events to the inner application, which is what makes it suitable for Django and other ASGI applications without lifespan support.

faust.contrib.fastapi.bind_to_running_loop(app: AppT) → Any

Bind app to the event loop that is currently running.

Returns the running loop.

Raises:

• RuntimeError – if there is no running event loop.

• LoopMismatch – if the app is already bound to a different loop.

Return type:

Any

faust.contrib.fastapi.faust_app_running(app: AppT, *, finalize: bool = True, discover: Optional[bool] = None, stop_timeout: Optional[float] = None) → AsyncIterator[AppT]

Start app on the running loop, and stop it on exit.

Use this when you have a lifespan of your own to compose with:

@asynccontextmanager
async def lifespan(api: FastAPI):
    async with faust_app_running(faust_app):
        ml_models["answer"] = load_model()
        yield
        ml_models.clear()

Parameters:

• app (AppT) – the Faust app to run.

• finalize (bool) – call finalize() before starting.

• discover (_UnionGenericAlias[bool, None]) – run autodiscovery. The default (None) discovers when the app is configured with autodiscover.

• stop_timeout (_UnionGenericAlias[float, None]) – seconds to wait for a graceful stop. None waits indefinitely.

The app is started with maybe_start(), so this composes with an app that is already running (and will not stop one it did not start).

Return type:

_GenericAlias[AppT]

faust.contrib.fastapi.faust_lifespan(app: AppT, *, opentelemetry: Optional[bool] = None, **kwargs: Any) → Callable[[...], Any]

Build an ASGI lifespan handler that runs app.

Accepts the same keyword arguments as faust_app_running():

api = FastAPI(lifespan=faust_lifespan(faust_app))

If OpenTelemetry is installed and configured, the ASGI application is instrumented automatically; pass opentelemetry=False to opt out.

Return type:

_CallableGenericAlias[…, Any]

faust.contrib.fastapi.serve_asgi(app: AppT, asgi_app: Any, *, host: Optional[str] = None, port: Optional[int] = None, **uvicorn_options: Any) → Type[AsgiService]

Use asgi_app as the Faust worker’s web application.

The ASGI server replaces the legacy faust.web aiohttp server. It starts once the app is up, after table recovery has finished, and obeys web_enabled and the worker’s --without-web option:

api = FastAPI()
serve_asgi(faust_app, api)

By default uvicorn binds to web_bind and web_port. @app.page and the built-in Faust aiohttp endpoints are not mounted; define routes with the selected framework instead.

Return type:

_GenericAlias[AsgiService]