FastAPI Plugin for FastStream

A plugin that allows you to use Depends and FastAPI other objects in the FastStream

Features

Use FastAPI Dependency Injection

In FastStream handlers, it will be possible to use the familiar DI from FastAPI

fromfastapiimportPath, Body, Header, DependsclassBodyModel(BaseModel):
field: int@broker.subscriber("subject.{num}")asyncdefsubscriber_handler(
num: Annotated[int, Path()],
body: Annotated[BodyModel, Body()],
x_user_id: Annotated[int, Header()],
my_dep: Annotated[int, Depends(int)],
) ->None:
...

Сan use FastAPI dependency overrides

fastapi=FastAPI()
fastapi.dependency_overrides[Dep] =lambda: "Dep"@broker.subscriber("subject")asyncdefsubscriber(dep: Annotated[str, De[]]) ->None:
assertdep=="Dep"

Using the FastStream Context

fromfaststreamimportContext@broker.subscriber("subject")asyncdefsubscriber_handler(context_data: Annotated[int, Context("data")]) ->Response: ...

Use FastAPI Request and Response

fromfastapiimportRequest, Responsefromfastapi.responsesimportJSONResponse@broker.subscriber("subject")asyncdefsubscriber_handler(request: Request) ->Response:
returnJSONResponse({"data": 1})

Minimalistic plugin connection to your application

All you need to do is wrap the FastAPI with the FastStreamAPI object from the plugin

application=FastStreamAPI(
NatsBroker(),
application=FastAPI(),
)
uvicorn.run(application)

Managing a lifespan state

Now the lifespan state is also available in FastStream handlers

@asynccontextmanagerasyncdeflifespan(app: FastAPI):
yield {"lifespan_data": "LIFESPAN DATA"}
@broker.subscriber("subject")asyncdefsubscriber(request: Request) ->None:
assertrequest.state.lifespan_data=="LIFESPAN DATA"

The ability to configure AsyncAPI

The ability to configure AsyncAPI using SpecificationFactory from FastStream and AsyncAPIConfig from the plugin

FastStreamAPI(
...,
specification=AsyncAPI(
title="My app",
version="1.0.0",
description="...",
...,
),
asyncapi_path="/fs_docs",
# orasyncapi_path=AsyncAPIRouter(
"/fs_docs",
description="...",
...
),
)

Backgrounds tasks

You can use BackgroundTasks from FastAPI in FastStream handlers

@broker.subscriber("subject")asyncdefhandler1(
tasks: BackgroundTasks,
) ->None:
tasks.add_task(...)