@@ -129,6 +129,49 @@ print(type(http_handler.from_json(customer_event_json)))
129129```
130130///
131131
132+ ### FastAPI
133+
134+ Both this package and [ FastAPI] ( https://fastapi.tiangolo.com/ ) are built on top
135+ of [ Pydantic] ( https://docs.pydantic.dev/latest/ ) . This means you don't need to instantiate
136+ a ` HTTPHandler ` to receive CloudEvents using a [ FastAPI] ( https://fastapi.tiangolo.com/ ) endpoint.
137+
138+ ``` python
139+ # ## Event classes omitted ###
140+
141+ Event = Annotated[
142+ Union[OrderCreatedEvent, CustomerCreatedEvent],
143+ Field(discriminator = " type" ),
144+ ]
145+
146+ # Endpoint for single events
147+ @router.post (" /event" , status_code = 204 )
148+ async def submit_event (
149+ event : Annotated[Event, Body()],
150+ content_type : Annotated[
151+ Literal[" application/cloudevents+json; charset=UTF-8" ], Header()
152+ ],
153+ ) -> None :
154+ do_something(event)
155+
156+ # Endpoint for event batches
157+ @router.post (" /batch" , status_code = 204 )
158+ async def submit_event_batch (
159+ event_batch : Annotated[List[Event], Body()],
160+ content_type : Annotated[
161+ Literal[" application/cloudevents-batch+json; charset=UTF-8" ], Header()
162+ ],
163+ ) -> None :
164+ for event in event_batch:
165+ do_something(event)
166+ ```
167+
168+ /// admonition | Generate the OpenAPI schema correctly
169+ type: tip
170+
171+ In order to have the OpenAPI spec correctly generated by FastAPI you'll need to
172+ work around some FastAPI limitations and manually specify some of the needed data.
173+ You can find a detailed example [ here] ( https://github.com/febus982/bootstrap-python-fastapi/blob/main/src/http_app/routes/events.py ) .
174+ ///
132175
133176## Serialize a JSON event
134177
0 commit comments