<!--suppress HtmlDeprecatedAttribute -->
<div align=center>
<h1>Cloud Eventful</h1>
<h3>Broker agnostic library to associate JSON Schemas to message broker topics.</h3>
<img src="https://img.shields.io/badge/License-MIT-blue.svg"
height="20"
alt="License: MIT">
<img src="https://img.shields.io/badge/code%20style-black-000000.svg"
height="20"
alt="Code style: black">
<img src="https://img.shields.io/pypi/v/cloudeventful.svg"
height="20"
alt="PyPI version">
<img src="https://img.shields.io/badge/coverage-100%25-success"
height="20"
alt="Code Coverage">
</div>
## Install
Cloud Eventful is on PyPI and can be installed with:
```shell
poetry add cloudeventful
```
or
```shell
pip install cloudeventful
```
## Usage
This library provides a `CloudEventful` class which can be used to generate
[CloudEvents](https://cloudevents.io/) and associate
[Pydantic](https://pydantic-docs.helpmanual.io/) models as the cloud event `data` field
on a per-topic basis.
### Model Registration
A model is associated with a pattern describing the topics it may be published to using
the `data_model` decorator.
```python
import re
from cloudeventful import CloudEventful
from pydantic import BaseModel
ce = CloudEventful(api_version="1.0.0", default_source="my/event/server")
@ce.data_model(re.compile(r"/.*/coffee"))
class Coffee(BaseModel):
flavor: str
```
### Cloud Event Generation
Once data models are registered, CloudEvent objects can be generated with an instance of
the generated model as the CloudEvent `data` property.
```pycon
>>> ce.event(Coffee(flavor="mocha"))
CloudEvent[ModelType](id='9b21a718-9dc1-4b56-a4ea-4e9911bc8ca6', source='my/event/server', specversion='1.0', type='Coffee', data=Coffee(flavor='mocha'), datacontenttype='application/json', dataschema='/Coffee', subject='Coffee', time=datetime.datetime(2022, 11, 19, 15, 33, 6, 39795))
```
### Publish
A publish function can be registered with a `CloudEventful` instance to enforce topic
integrity at run time. This is done by setting the `publish_function` property on a
`CloudEventful` instance.
A publish function must accept at least a topic arg as a str and a data arg as a
registered data model.
Then, the `CloudEventful` publish function can be used to wrap data models in a
CloudEvent and publish them as JSON strings. Keyword args will be passed to the
registered publish function.
## Example using MQTT with Paho
```python
import re
from cloudeventful import CloudEventful
import paho.mqtt.client as mqtt
from pydantic import BaseModel
server_id = "my/event/server"
client = mqtt.Client(server_id)
client.connect("127.0.0.1")
ce = CloudEventful(
api_version="1.0.0",
default_source=server_id,
publish_function=client.publish,
default_topic_factory=lambda m: f"/api/v1/{type(m).__name__.lower()}"
)
@ce.data_model(re.compile(r"/.*/coffee"))
class Coffee(BaseModel):
flavor: str
@ce.data_model(re.compile(r"/.*/pen"))
class Pen(BaseModel):
color: str
# Publish a data model wrapped in a cloud event.
ce.publish(Coffee(flavor="mocha"))
# Raise `ValueError` because topic does not match pattern of this model.
ce.publish(Pen(color="black"), topic="wrong-topic")
```
## Support The Developer
<a href="https://www.buymeacoffee.com/mburkard" target="_blank">
<img src="https://cdn.buymeacoffee.com/buttons/v2/default-blue.png"
width="217"
height="60"
alt="Buy Me A Coffee">
</a>
Raw data
{
"_id": null,
"home_page": "https://gitlab.com/mburkard/cloud-eventful",
"name": "cloudeventful",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.9,<4.0",
"maintainer_email": "",
"keywords": "",
"author": "Matthew Burkard",
"author_email": "matthewjburkard@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/29/59/fb843fcbae50dafe0cb1167575891944480ac455fe1844a045d1cb05cca0/cloudeventful-3.0.0.tar.gz",
"platform": null,
"description": "<!--suppress HtmlDeprecatedAttribute -->\n<div align=center>\n <h1>Cloud Eventful</h1>\n <h3>Broker agnostic library to associate JSON Schemas to message broker topics.</h3>\n <img src=\"https://img.shields.io/badge/License-MIT-blue.svg\"\n height=\"20\"\n alt=\"License: MIT\">\n <img src=\"https://img.shields.io/badge/code%20style-black-000000.svg\"\n height=\"20\"\n alt=\"Code style: black\">\n <img src=\"https://img.shields.io/pypi/v/cloudeventful.svg\"\n height=\"20\"\n alt=\"PyPI version\">\n <img src=\"https://img.shields.io/badge/coverage-100%25-success\"\n height=\"20\"\n alt=\"Code Coverage\">\n</div>\n\n## Install\n\nCloud Eventful is on PyPI and can be installed with:\n\n```shell\npoetry add cloudeventful\n```\n\nor\n\n```shell\npip install cloudeventful\n```\n\n## Usage\n\nThis library provides a `CloudEventful` class which can be used to generate\n[CloudEvents](https://cloudevents.io/) and associate\n[Pydantic](https://pydantic-docs.helpmanual.io/) models as the cloud event `data` field\non a per-topic basis.\n\n### Model Registration\n\nA model is associated with a pattern describing the topics it may be published to using\nthe `data_model` decorator.\n\n```python\nimport re\n\nfrom cloudeventful import CloudEventful\nfrom pydantic import BaseModel\n\nce = CloudEventful(api_version=\"1.0.0\", default_source=\"my/event/server\")\n\n\n@ce.data_model(re.compile(r\"/.*/coffee\"))\nclass Coffee(BaseModel):\n flavor: str\n```\n\n### Cloud Event Generation\n\nOnce data models are registered, CloudEvent objects can be generated with an instance of\nthe generated model as the CloudEvent `data` property.\n\n```pycon\n>>> ce.event(Coffee(flavor=\"mocha\"))\nCloudEvent[ModelType](id='9b21a718-9dc1-4b56-a4ea-4e9911bc8ca6', source='my/event/server', specversion='1.0', type='Coffee', data=Coffee(flavor='mocha'), datacontenttype='application/json', dataschema='/Coffee', subject='Coffee', time=datetime.datetime(2022, 11, 19, 15, 33, 6, 39795))\n```\n\n### Publish\n\nA publish function can be registered with a `CloudEventful` instance to enforce topic\nintegrity at run time. This is done by setting the `publish_function` property on a\n`CloudEventful` instance.\n\nA publish function must accept at least a topic arg as a str and a data arg as a\nregistered data model.\n\nThen, the `CloudEventful` publish function can be used to wrap data models in a\nCloudEvent and publish them as JSON strings. Keyword args will be passed to the\nregistered publish function.\n\n## Example using MQTT with Paho\n\n```python\nimport re\n\nfrom cloudeventful import CloudEventful\nimport paho.mqtt.client as mqtt\nfrom pydantic import BaseModel\n\nserver_id = \"my/event/server\"\n\nclient = mqtt.Client(server_id)\nclient.connect(\"127.0.0.1\")\n\nce = CloudEventful(\n api_version=\"1.0.0\",\n default_source=server_id,\n publish_function=client.publish,\n default_topic_factory=lambda m: f\"/api/v1/{type(m).__name__.lower()}\"\n)\n\n\n@ce.data_model(re.compile(r\"/.*/coffee\"))\nclass Coffee(BaseModel):\n flavor: str\n\n\n@ce.data_model(re.compile(r\"/.*/pen\"))\nclass Pen(BaseModel):\n color: str\n\n\n# Publish a data model wrapped in a cloud event.\nce.publish(Coffee(flavor=\"mocha\"))\n# Raise `ValueError` because topic does not match pattern of this model.\nce.publish(Pen(color=\"black\"), topic=\"wrong-topic\")\n```\n\n## Support The Developer\n\n<a href=\"https://www.buymeacoffee.com/mburkard\" target=\"_blank\">\n <img src=\"https://cdn.buymeacoffee.com/buttons/v2/default-blue.png\"\n width=\"217\"\n height=\"60\"\n alt=\"Buy Me A Coffee\">\n</a>\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Broker agnostic library to associate JSON Schemas to message broker topics.",
"version": "3.0.0",
"project_urls": {
"Homepage": "https://gitlab.com/mburkard/cloud-eventful",
"Repository": "https://gitlab.com/mburkard/cloud-eventful"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "ece166671706cb912a1e9b08c1d6bf911617548587fae8dcdaa110708ac5ab4a",
"md5": "d7f424d97083ec22065f4ddb22b61991",
"sha256": "e5a97d7d44819e24de95027ec8534d7768a2a51644f67542acaacef7e43fd99b"
},
"downloads": -1,
"filename": "cloudeventful-3.0.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "d7f424d97083ec22065f4ddb22b61991",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9,<4.0",
"size": 7804,
"upload_time": "2023-11-24T19:09:43",
"upload_time_iso_8601": "2023-11-24T19:09:43.921589Z",
"url": "https://files.pythonhosted.org/packages/ec/e1/66671706cb912a1e9b08c1d6bf911617548587fae8dcdaa110708ac5ab4a/cloudeventful-3.0.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "2959fb843fcbae50dafe0cb1167575891944480ac455fe1844a045d1cb05cca0",
"md5": "5e52e2774d04a554d681e2f4834b89a8",
"sha256": "a93fbeb4e2af469b1c651b7e34e5ce380ff807c325e73d39d1220ec85d25fbf1"
},
"downloads": -1,
"filename": "cloudeventful-3.0.0.tar.gz",
"has_sig": false,
"md5_digest": "5e52e2774d04a554d681e2f4834b89a8",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9,<4.0",
"size": 7746,
"upload_time": "2023-11-24T19:09:45",
"upload_time_iso_8601": "2023-11-24T19:09:45.534230Z",
"url": "https://files.pythonhosted.org/packages/29/59/fb843fcbae50dafe0cb1167575891944480ac455fe1844a045d1cb05cca0/cloudeventful-3.0.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-11-24 19:09:45",
"github": false,
"gitlab": true,
"bitbucket": false,
"codeberg": false,
"gitlab_user": "mburkard",
"gitlab_project": "cloud-eventful",
"lcname": "cloudeventful"
}
JSON