# cloudevents-pydantic
[](https://pypi.org/project/cloudevents-pydantic/)
[](https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md#work-in-progress)
[](https://github.com/febus982/cloudevents-pydantic/actions/workflows/python-tests.yml)
[](https://codeclimate.com/github/febus982/cloudevents-pydantic/maintainability)
[](https://codeclimate.com/github/febus982/cloudevents-pydantic/test_coverage)
[](https://mypy-lang.org/)
[](https://github.com/charliermarsh/ruff)
[](https://github.com/psf/black)
[](https://github.com/PyCQA/bandit)
This is an implementation of the [CloudEvents spec](https://github.com/cloudevents/spec/tree/main) using
[Pydantic V2](https://docs.pydantic.dev/latest/) for high performance during validation and serialization.
It is meant to support natively [FastAPI](https://fastapi.tiangolo.com/)
and [FastStream](https://faststream.airt.ai/latest/) (WIP)
Currently supported bindings:
| Binding | Format | Single | Batch |
|---------|:-------|:-------:|:-------:|
| HTTP | JSON | ✅ | ✅ |
| HTTP | Binary | planned | planned |
| KAFKA | JSON | planned | planned |
| KAFKA | Binary | planned | planned |
## How to use
```shell
pip install cloudevents-pydantic
```
```python
from cloudevents_pydantic.bindings.http import HTTPHandler
from cloudevents_pydantic.events import CloudEvent
handler = HTTPHandler()
minimal_attributes = {
"type": "com.example.string",
"source": "https://example.com/event-producer",
}
# `CloudEvent` is a Pydantic model to handle validation and serialization
# `event_factory` is a helper method to autogenerate some of the mandatory
# such as id, time, specversion
event = CloudEvent.event_factory(**minimal_attributes)
# Single event HTTP serialization
headers, single_event_as_json = handler.to_json(event)
# Batch of events HTTP serialization
headers, batch_of_events_as_json = handler.to_json_batch([event])
# Parsing a JSON string for a single event
parsed_event = handler.from_json(single_event_as_json)
# Parsing a JSON string for a single event
parsed_event_list = handler.from_json(batch_of_events_as_json)
```
Refer to the [docs](https://febus982.github.io/cloudevents-pydantic/) for more advanced use cases and
for details on how to create custom events.
## Performance
Using pydantic gives a great performance boost if compared to the official SDK. (there's obviously
some performance issue in the official serialization using pydantic)
These results come from a Macbook Pro M3 Max on python 3.12. Feel free to run the `benchmark.py`
script yourself.
```
Timings for HTTP JSON deserialization:
This package: 3.0855846670019673
Official SDK with pydantic model: 15.35431600001175
Official SDK with http model: 13.728038166998886
Timings for HTTP JSON serialization:
This package: 4.292417042001034
Official SDK with pydantic model: 44.50933354199515
Official SDK with http model: 8.929204874992138
```
## Commands for development
All the common commands used during development can be run using make targets:
* `make dev-dependencies`: Install dev requirements
* `make update-dependencies`: Update dev requirements
* `make fix`: Run code style and lint automatic fixes (where possible)
* `make test`: Run test suite against system python version
* `make check`: Run tests against all available python versions, code style and lint checks
* `make type`, `make format`, `make lint`, `make bandit`: Run the relevant check
* `make docs`: Render the mkdocs website locally
Raw data
{
"_id": null,
"home_page": "https://febus982.github.io/cloudevents-pydantic",
"name": "cloudevents-pydantic",
"maintainer": null,
"docs_url": null,
"requires_python": "<3.14,>=3.9",
"maintainer_email": null,
"keywords": null,
"author": "Federico Busetti",
"author_email": "729029+febus982@users.noreply.github.com",
"download_url": "https://files.pythonhosted.org/packages/cf/e4/cc005b07e1b2f01be0f6058a5ef747dcc5a91f77d217c0f2a25442eceba4/cloudevents_pydantic-0.0.3.tar.gz",
"platform": null,
"description": "# cloudevents-pydantic\n\n[](https://pypi.org/project/cloudevents-pydantic/)\n[](https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md#work-in-progress)\n\n[](https://github.com/febus982/cloudevents-pydantic/actions/workflows/python-tests.yml)\n[](https://codeclimate.com/github/febus982/cloudevents-pydantic/maintainability)\n[](https://codeclimate.com/github/febus982/cloudevents-pydantic/test_coverage)\n\n[](https://mypy-lang.org/)\n[](https://github.com/charliermarsh/ruff)\n[](https://github.com/psf/black)\n[](https://github.com/PyCQA/bandit)\n\nThis is an implementation of the [CloudEvents spec](https://github.com/cloudevents/spec/tree/main) using\n[Pydantic V2](https://docs.pydantic.dev/latest/) for high performance during validation and serialization.\n\nIt is meant to support natively [FastAPI](https://fastapi.tiangolo.com/)\nand [FastStream](https://faststream.airt.ai/latest/) (WIP)\n\nCurrently supported bindings:\n\n| Binding | Format | Single | Batch |\n|---------|:-------|:-------:|:-------:|\n| HTTP | JSON | \u2705 | \u2705 |\n| HTTP | Binary | planned | planned |\n| KAFKA | JSON | planned | planned |\n| KAFKA | Binary | planned | planned |\n\n## How to use\n\n```shell\npip install cloudevents-pydantic\n```\n\n```python\nfrom cloudevents_pydantic.bindings.http import HTTPHandler\nfrom cloudevents_pydantic.events import CloudEvent\n\nhandler = HTTPHandler()\nminimal_attributes = {\n \"type\": \"com.example.string\",\n \"source\": \"https://example.com/event-producer\",\n}\n\n# `CloudEvent` is a Pydantic model to handle validation and serialization\n# `event_factory` is a helper method to autogenerate some of the mandatory \n# such as id, time, specversion\nevent = CloudEvent.event_factory(**minimal_attributes)\n\n# Single event HTTP serialization\nheaders, single_event_as_json = handler.to_json(event)\n\n# Batch of events HTTP serialization\nheaders, batch_of_events_as_json = handler.to_json_batch([event])\n\n# Parsing a JSON string for a single event\nparsed_event = handler.from_json(single_event_as_json)\n\n# Parsing a JSON string for a single event\nparsed_event_list = handler.from_json(batch_of_events_as_json)\n```\n\nRefer to the [docs](https://febus982.github.io/cloudevents-pydantic/) for more advanced use cases and\nfor details on how to create custom events.\n\n## Performance\n\nUsing pydantic gives a great performance boost if compared to the official SDK. (there's obviously\nsome performance issue in the official serialization using pydantic)\n\nThese results come from a Macbook Pro M3 Max on python 3.12. Feel free to run the `benchmark.py`\nscript yourself.\n\n```\nTimings for HTTP JSON deserialization:\nThis package: 3.0855846670019673\nOfficial SDK with pydantic model: 15.35431600001175\nOfficial SDK with http model: 13.728038166998886\n\nTimings for HTTP JSON serialization:\nThis package: 4.292417042001034\nOfficial SDK with pydantic model: 44.50933354199515\nOfficial SDK with http model: 8.929204874992138\n```\n\n\n## Commands for development\n\nAll the common commands used during development can be run using make targets:\n\n* `make dev-dependencies`: Install dev requirements\n* `make update-dependencies`: Update dev requirements\n* `make fix`: Run code style and lint automatic fixes (where possible)\n* `make test`: Run test suite against system python version\n* `make check`: Run tests against all available python versions, code style and lint checks\n* `make type`, `make format`, `make lint`, `make bandit`: Run the relevant check\n* `make docs`: Render the mkdocs website locally\n\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "An implementation of the CloudEvents spec using Pydantic V2",
"version": "0.0.3",
"project_urls": {
"Homepage": "https://febus982.github.io/cloudevents-pydantic",
"Repository": "https://github.com/febus982/cloudevents-pydantic"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "39695ed8a9857d4b3a3db2e30b94aa60c95220af63b631c4b10462759f39d5c8",
"md5": "a9e933031697ca15f6baff7970ac9638",
"sha256": "7d75b9e9f0942174c0843e6322d18974c57532274769ccb80677bf07a9e5d5c9"
},
"downloads": -1,
"filename": "cloudevents_pydantic-0.0.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "a9e933031697ca15f6baff7970ac9638",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<3.14,>=3.9",
"size": 19997,
"upload_time": "2024-12-15T16:04:03",
"upload_time_iso_8601": "2024-12-15T16:04:03.097550Z",
"url": "https://files.pythonhosted.org/packages/39/69/5ed8a9857d4b3a3db2e30b94aa60c95220af63b631c4b10462759f39d5c8/cloudevents_pydantic-0.0.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "cfe4cc005b07e1b2f01be0f6058a5ef747dcc5a91f77d217c0f2a25442eceba4",
"md5": "c647dc3cb6196db73bdbf290589fe19b",
"sha256": "5f20d479ae18094b2b0b7f2072cc1b615931af788f0c9ae35bcdfd1001c2aee1"
},
"downloads": -1,
"filename": "cloudevents_pydantic-0.0.3.tar.gz",
"has_sig": false,
"md5_digest": "c647dc3cb6196db73bdbf290589fe19b",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<3.14,>=3.9",
"size": 11151,
"upload_time": "2024-12-15T16:04:04",
"upload_time_iso_8601": "2024-12-15T16:04:04.590578Z",
"url": "https://files.pythonhosted.org/packages/cf/e4/cc005b07e1b2f01be0f6058a5ef747dcc5a91f77d217c0f2a25442eceba4/cloudevents_pydantic-0.0.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-15 16:04:04",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "febus982",
"github_project": "cloudevents-pydantic",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"tox": true,
"lcname": "cloudevents-pydantic"
}
JSON
