FastEDA

Like FastAPI but for event processing

Current status: WIP

Example of usage

import pydantic
from fasteda import adapter, app


apps = app.FastEDA(adapter=adapter.pydantic, middlewares=[])

class Client(pydantic.BaseModel):
    id: int
    name: str

@apps.add("client.create.v1")
def create_client(client: Client) -> None:
    print(f"Client {client.id} created")  # noqa: T201

Run consumer:

> python -m fasteda.consumer --app examples.dlq_v2.apps --bootstrap_servers localhost:9092 --group_id my-app

See more examples:

• Parse JWT here
• Parse databus here
• DLQ Mock Producer here
• DLQ Full here
• Lifespan here
• Router here