Event / AMQP Integration?

[a-n-d-i]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

print("Maybe I should ask this question in the chat?")

# This is how it could look like
@app.event("someeventname")
function handle_event(message: some_model):
     print("Event handled")

Description

Question: I’m a big fan of FastAPI and Nameko, a Microservice Framework which is extremly handy for event based stuff with brokers, especially AMQP. You just annotate a function with @event_handler and it creates exchanges, queues, ..., and sets up routing. In my opinon event handling is quite different to distributed task queues such as Celery. Sadly, Nameko is bit dated, it’s based on eventlet and it was build long before type hints.

So, I’m wondering: Has anybody ever tried to combine these two concepts or is interested in working on it? I’m contemplating to give it a stab and would love to hear some thoughts about...

Cheers, Andi

Operating System

Linux

Operating System Details

No response

FastAPI Version

0.70.1

Python Version

3.9

Additional Context

No response

[Lancetnik]

Hi! I tried implement this feature as a part of FastAPI app, but it requires a lot of core changes.
Cuz I created a little project Propan with implementation of major FastAPI features: pydantic type casting and dependencies injection.
You can use it as core of your RabbitMQ/Nats based applications or as a part of your FastAPI app following way:

from fastapi import FastAPI
from propan.brokers import RabbitBroker

broker = RabbitBroker("amqp://guest:guest@localhost:5672/")

app = FastAPI()

@broker.handle("test")
async def base_handler(body):
    print(body)

@app.on_event("startup")
async def start_broker():
    await broker.start()

@app.on_event("shutdown")
async def stop_broker():
    await broker.close()

This project just created and actively grows everyday. As a creator I am opened to all thoughts about features to implement. Also I was glad to new contributors or any feedback about this project.

[YuriiMotov]

Now it's FastStream