Metadata-Version: 2.4
Name: faststream_fastapi
Version: 1.0.0
Summary: FastAPI integration for FastStream
Keywords: rabbitmq,kafka,nats,redis,mqtt,asyncapi,framework,message brokers,fastapi
Author: Ivan Kirpichnikov
Author-email: Ivan Kirpichnikov <mmssvvvv570@gmail.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 5 - Production/Stable
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Operating System :: OS Independent
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development
Classifier: Topic :: System :: Networking
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Information Technology
Classifier: Intended Audience :: System Administrators
Classifier: Environment :: Web Environment
Classifier: Framework :: AsyncIO
Classifier: Framework :: Pydantic
Classifier: Framework :: Pydantic :: 1
Classifier: Framework :: Pydantic :: 2
Requires-Dist: fastapi<1.0.0
Requires-Dist: faststream>=0.7.0
Requires-Python: >=3.10
Description-Content-Type: text/markdown

# 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**
```py
from fastapi import Path, Body, Header, Depends

class BodyModel(BaseModel):
    field: int

@broker.subscriber("subject.{num}")
async def subscriber_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
```py
fastapi = FastAPI()
fastapi.dependency_overrides[Dep] = lambda: "Dep"

@broker.subscriber("subject")
async def subscriber(dep: Annotated[str, De[]]) -> None:
    assert dep == "Dep"
```

### Using the FastStream Context
```py
from faststream import Context

@broker.subscriber("subject")
async def subscriber_handler(context_data: Annotated[int, Context("data")]) -> Response: ...
```

### Use FastAPI Request and Response
```py
from fastapi import Request, Response
from fastapi.responses import JSONResponse

@broker.subscriber("subject")
async def subscriber_handler(request: Request) -> Response:
    return JSONResponse({"data": 1})
```

### Minimalistic plugin connection to your application

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

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

### Managing a lifespan state
Now the **lifespan state** is also available in **FastStream handlers**

```py
@asynccontextmanager
async def lifespan(app: FastAPI):
    yield {"lifespan_data": "LIFESPAN DATA"}

@broker.subscriber("subject")
async def subscriber(request: Request) -> None:
    assert request.state.lifespan_data == "LIFESPAN DATA"
```

### The ability to configure AsyncAPI

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

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

### Backgrounds tasks

You can use **BackgroundTasks** from **FastAPI** in **FastStream handlers**

```py
@broker.subscriber("subject")
async def handler1(
    tasks: BackgroundTasks,
) -> None:
    tasks.add_task(...)
```
