FastAPI CloudEvents

raw JSON →

2.0.2 verified Fri May 01 auth: no python

A FastAPI plugin for integrating CloudEvents. Current version 2.0.2, release cadence: sporadic.

pip install fastapi-cloudevents

Common errors

error ImportError: cannot import name 'CloudEvent' from 'fastapi_cloudevents' ↓

cause CloudEvent class is not part of fastapi-cloudevents; it comes from the cloudevents SDK.

fix

Install cloudevents and import CloudEvent from cloudevents.http.

error AttributeError: 'Request' object has no attribute 'body' ↓

cause Trying to access `.body` on a Request that hasn't been read yet (async context).

fix

Use await request.body() instead of request.body.

Warnings

breaking Version 2.0.0 introduced Pydantic V2 support, dropping V1. If your project uses Pydantic V1, stay on v1.x or migrate models. ↓

fix Use Pydantic V2 or pin fastapi-cloudevents<2.0.0.

deprecated Use of `from_event` on raw dicts is deprecated; always pass a FastAPI Request object. ↓

fix Pass the `Request` object from FastAPI endpoint, not a dict.

gotcha The `to_event` function returns a Starlette Response, not a CloudEvent object. Do not try to call `.data` on it. ↓

fix Return the response directly from your endpoint, or inspect the body via `response.body`.

Imports

CloudEvent
wrong
```
from fastapi_cloudevents import CloudEvent
```
correct
```
from cloudevents.http import CloudEvent
```
CloudEvent data model is from cloudevents SDK, not from fastapi-cloudevents.

from_event

from fastapi_cloudevents import from_event

to_event

from fastapi_cloudevents import to_event

Quickstart

Basic example of receiving and sending CloudEvents with FastAPI.

from fastapi import FastAPI, Request
from fastapi_cloudevents import from_event, to_event
from cloudevents.http import CloudEvent

app = FastAPI()

@app.post("/")
async def receive(request: Request):
    event = from_event(request)
    print(event)
    return to_event(CloudEvent({"type": "example", "source": "test"}, {"message": "Hello"}))