faust.contrib.asgi

Run Faust with any ASGI application or web framework service.

Faust and an ASGI server must share a single event loop: a Faust producer created on one loop cannot be awaited from another. This module provides the two directions that need:

1. The ASGI server drives. Your own FastAPI application is served by uvicorn, and Faust is started and stopped from its lifespan:

   import faust
   from fastapi import FastAPI
   from faust.contrib.asgi import faust_lifespan
   
   faust_app = faust.App("hello", broker="kafka://localhost:9092")
   greetings = faust_app.topic("greetings", value_type=str)
   
   api = FastAPI(lifespan=faust_lifespan(faust_app))
   
   @api.post("/greet")
   async def greet(text: str):
       await greetings.send(value=text)
       return {"ok": True}
   

   Run it with uvicorn myapp:api.

2. The Faust worker drives. faust worker runs as usual and your ASGI application is served from inside it, on the worker’s own loop:

   from faust.contrib.asgi import serve_asgi
   
   api = FastAPI()
   serve_asgi(faust_app, api, port=8000)
   

   Run it with faust -A myapp worker -l info.

Nothing here imports https://pypi.org/project/fastapi/ or https://pypi.org/project/starlette/, so it works with FastAPI, Starlette, Quart, Litestar, Django ASGI, or any other ASGI callable. For frameworks that are not ASGI-based, register a mode.Service directly with faust.App.web_server().

Install the optional ASGI server with pip install faust-streaming[asgi].

exception faust.contrib.asgi.LoopMismatch

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

class faust.contrib.asgi.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)>

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.asgi.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.asgi.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.asgi.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.asgi.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.asgi.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]