Integration with FastAPI problems

[xKlask]

To begin with, I would like to say a great thanks to the developer, Propan is magnificent. Very long thought to simplify the work and bring in a convenient format, but never got it together.

I had some problems with the integration with FastAPI, probably due to incomplete documentation or other things.

FastAPI 0.96.0
Propan 0.1.2.12

1. If we consider a minimal application from documentation to send event:

from fastapi import FastAPI
from propan.fastapi import RabbitRouter
import uvicorn

app = FastAPI()

router = RabbitRouter("amqp://guest:guest@localhost:5672")


@router.get("/")
async def hello_http():
    await router.broker.publish("Hello, Rabbit!", "test")
    return "Hello, HTTP!"

app.include_router(router)

if __name__ == '__main__':
    uvicorn.run(app)

We get the following error:

line 12, in hello_http    await router.broker.publish("Hello, Rabbit!", "test")
ValueError: RabbitBroker channel not started yet

This is because the "lifespan" is not cast to the router, but can only be put at the stage of creating the application. Unlike the deprecated "on_startup".
We fix this by adding a router lifespan at the stage of creating the application:

router = RabbitRouter("amqp://guest:guest@localhost:5672")
app = FastAPI(lifespan=router.lifespan_context)

or send all events like example from tests:

async with router.lifespan_context(None):
    await router.broker.publish("Hello, Rabbit!", "test")

2. Receiving messages. Let's consider a minimal variant:

from propan.fastapi import RabbitRouter
from pydantic import BaseModel
from fastapi import FastAPI
import uvicorn

router = RabbitRouter("amqp://guest:guest@localhost:5672")
app = FastAPI(lifespan=router.lifespan_context)


class Message(BaseModel):
    string: str


@router.get('/')
async def hello_http():
    await router.broker.publish(Message(string='test'), "test")
    return 'Hello, HTTP!'


@router.event('test')
async def hello(m: Message):
    return True

app.include_router(router)

if __name__ == '__main__':
    uvicorn.run(app)

We get a pydantic validation error because we take a byte message and don't parse it.
Because the file "propan/fastapi/core/route.py" on line 35 has hardcode _raw=True

broker.handle(path, _raw=True, **handle_kwargs)(handler)

Since we don't convert bytes to dict, pydantic can't validate (which would be very convenient)
changing the _raw flag to False doesn't fix it, because the app method on line 75 is waiting for a "NativeMessage" object

async def app(message: NativeMessage) -> Any:

But the decode method only passes the dict contained in the .body

My crappy fix for the situation is _raw=False and add a check in the app method:

async def app(message: Union[NativeMessage, AnyDict]) -> Any:
    if isinstance(message, dict):
        body = message
        headers = {}
    else:
        body: Union[AnyDict, bytes] = message.body
        headers = message.headers
    if first_arg is not None:
        if not isinstance(body, dict):  # pragma: no branch
            body = {first_arg: body}

        session = cls(body, headers)
    else:
        session = cls()
    return await func(session)

Hopefully we can find a good solution for easy integration with FastAPI

[Lancetnik]

@xKlask
Wow, thanks for your so detail Issue and so pleasure feedback!
Looks like I broke the FastAPI part behavior at 0.1.2.6 release and tests passes it. Now you can a little downgrade to previous version or stay with your changes. I will try to fix it soon: today/tomorrow.

[Lancetnik]

@xKlask great job again! I fix serialization Issue from Propan side, but ignoring router lifespan from FastAPI side is a bug of that framework. I fix it too and will try to PR it to FastAPI.

[Lancetnik]

@xKlask with 0.1.2.13 version I fixed messages serialization.
Anfortunately, you should use until FastAPI updates this part...

router = RabbitRouter("amqp://guest:guest@localhost:5672")
app = FastAPI(lifespan=router.lifespan_context)

[Lancetnik]

@xKlask can you support this PR?

[xKlask]

@Lancetnik can you read email, pls?