docs/protocol_bindings/http_binding.md
# HTTP binding
Using the HTTP binding handler is straightforward. Just remember to reuse the same
`HTTPHandler` multiple times.
/// tab | ✅ Good
```python
from cloudevents_pydantic.events import CloudEvent
from cloudevents_pydantic.bindings.http import HTTPHandler
class OrderCreated(CloudEvent):
...
http_handler = HTTPHandler(OrderCreated)
def do_something():
http_handler.from_json("json_body")
```
///
/// tab | ❌ Bad
```python
from cloudevents_pydantic.events import CloudEvent
from cloudevents_pydantic.bindings.http import HTTPHandler
class OrderCreated(CloudEvent):
...
def do_something():
http_handler = HTTPHandler(OrderCreated)
http_handler.from_json("json_body")
```
///
/// admonition | Why you have to reuse the same object?
type: tip
When the HTTPHandler instance is created it creates internally instances of Pydantic `TypeAdapter`
for the event class, to handle efficiently event serialization and discriminated unions. This is
an expensive operation. Check the [Pydantic documentation](https://docs.pydantic.dev/latest/concepts/performance/#typeadapter-instantiated-once) about this.
///
## Deserialize a JSON event
HTTP deserialization parses the body to reconstruct the event.
/// tab | Custom Event class
```python
from cloudevents_pydantic.events import CloudEvent
from cloudevents_pydantic.bindings.http import HTTPHandler
class OrderCreated(CloudEvent):
...
single_event_json = '{"data":null,"source":"https://example.com/event-producer","id":"b96267e2-87be-4f7a-b87c-82f64360d954","type":"com.example.string","specversion":"1.0","time":"2022-07-16T12:03:20.519216+04:00","subject":null,"datacontenttype":null,"dataschema":null}'
batch_event_json = '[{"data":null,"source":"https://example.com/event-producer","id":"b96267e2-87be-4f7a-b87c-82f64360d954","type":"com.example.string","specversion":"1.0","time":"2022-07-16T12:03:20.519216+04:00","subject":null,"datacontenttype":null,"dataschema":null}]'
http_handler = HTTPHandler(OrderCreated)
# Single event
event = http_handler.from_json(single_event_json)
# Batch (list) of events
batch_of_events = http_handler.from_json_batch(batch_event_json)
```
///
/// tab | CloudEvent class
```python
from cloudevents_pydantic.events import CloudEvent
from cloudevents_pydantic.bindings.http import HTTPHandler
minimal_attributes = {
"type": "order_created",
"source": "https://example.com/event-producer",
"id": "b96267e2-87be-4f7a-b87c-82f64360d954",
"specversion": "1.0",
}
http_handler = HTTPHandler()
event = CloudEvent.event_factory(**minimal_attributes)
# Single event
event = http_handler.to_json(event)
# Batch (list) of events
batch_of_events = http_handler.to_json_batch([event])
```
///
/// details | Use discriminated Unions to handle multiple Event classes
type: warning
You'll want to use [discriminated unions](https://docs.pydantic.dev/latest/concepts/unions/#discriminated-unions)
as event class and use a single `HTTPHandler` for multiple Event classes to be more efficient on validation
and to produce a correct schema.
```python
from typing import Annotated, Literal, Union
from pydantic import Field
from typing_extensions import TypedDict
from cloudevents_pydantic.bindings.http import HTTPHandler
from cloudevents_pydantic.events import CloudEvent
class OrderCreatedEvent(CloudEvent):
data: TypedDict("OrderCreatedData", {"order_id": str})
type: Literal["order_created"]
class CustomerCreatedEvent(CloudEvent):
data: TypedDict("CustomerCreatedData", {"customer_id": str})
type: Literal["customer_created"]
Event = Annotated[
Union[OrderCreatedEvent, CustomerCreatedEvent],
Field(discriminator="type"),
]
http_handler = HTTPHandler(Event)
customer_event_json = '{"data":{"customer_id":"123"},"source":"customer_service","id":"123","type":"customer_created","specversion":"1.0","time":null,"subject":null,"datacontenttype":null,"dataschema":null}'
print(type(http_handler.from_json(customer_event_json)))
# <class '__main__.CustomerCreatedEvent'>
```
///
### FastAPI
Both this package and [FastAPI](https://fastapi.tiangolo.com/) are built on top
of [Pydantic](https://docs.pydantic.dev/latest/). This means you don't need to instantiate
a `HTTPHandler` to receive CloudEvents using a [FastAPI](https://fastapi.tiangolo.com/) endpoint.
```python
### Event classes omitted ###
Event = Annotated[
Union[OrderCreatedEvent, CustomerCreatedEvent],
Field(discriminator="type"),
]
# Endpoint for single events
@router.post("/event", status_code=204)
async def submit_event(
event: Annotated[Event, Body()],
content_type: Annotated[
Literal["application/cloudevents+json; charset=UTF-8"], Header()
],
) -> None:
do_something(event)
# Endpoint for event batches
@router.post("/batch", status_code=204)
async def submit_event_batch(
event_batch: Annotated[List[Event], Body()],
content_type: Annotated[
Literal["application/cloudevents-batch+json; charset=UTF-8"], Header()
],
) -> None:
for event in event_batch:
do_something(event)
```
/// admonition | Generate the OpenAPI schema correctly
type: tip
In order to have the OpenAPI spec correctly generated by FastAPI you'll need to
work around some FastAPI limitations and manually specify some of the needed data.
You can find a detailed example [here](https://github.com/febus982/bootstrap-python-fastapi/blob/main/src/http_app/routes/events.py).
///
## Serialize a JSON event
HTTP serialization returns header and body to be used in a HTTP request.
/// tab | Custom Event class
```python
from cloudevents_pydantic.events import CloudEvent
from cloudevents_pydantic.bindings.http import HTTPHandler
class OrderCreated(CloudEvent):
...
minimal_attributes = {
"type": "order_created",
"source": "https://example.com/event-producer",
"id": "b96267e2-87be-4f7a-b87c-82f64360d954",
"specversion": "1.0",
}
http_handler = HTTPHandler(OrderCreated)
event = OrderCreated.event_factory(**minimal_attributes)
# Single event
headers, body = http_handler.to_json(event)
# Batch (list) of events
headers, body = http_handler.to_json_batch([event])
```
///
/// tab | CloudEvent class
```python
from cloudevents_pydantic.events import CloudEvent
from cloudevents_pydantic.bindings.http import HTTPHandler
minimal_attributes = {
"type": "order_created",
"source": "https://example.com/event-producer",
"id": "b96267e2-87be-4f7a-b87c-82f64360d954",
"specversion": "1.0",
}
http_handler = HTTPHandler()
event = CloudEvent.event_factory(**minimal_attributes)
# Single event
json_string = http_handler.to_json(event)
# Batch (list) of events
json_batch_string = http_handler.to_json_batch([event])
```
///