[](https://github.com/camunda-community-hub/community)
[](https://github.com/Camunda-Community-Hub/community/blob/main/extension-lifecycle.md#stable-)
[](https://coveralls.io/github/JonatanMartens/pyzeebe?branch=master)
[](https://pyzeebe.readthedocs.io/en/latest/?badge=stable)
# Pyzeebe
pyzeebe is a python grpc client for Zeebe.
Zeebe version support:
| Pyzeebe version | Tested Zeebe versions |
| :-------------: | ---------------------- |
| 3.x.x | 1.0.0 |
| 2.x.x | 0.23, 0.24, 0.25, 0.26 |
| 1.x.x | 0.23, 0.24 |
## Getting Started
To install:
`pip install pyzeebe`
For full documentation please visit: https://pyzeebe.readthedocs.io/en/stable/
## Usage
### Worker
The `ZeebeWorker` class gets jobs from the gateway and runs them.
```python
import asyncio
from pyzeebe import ZeebeWorker, Job, create_insecure_channel
channel = create_insecure_channel(hostname="localhost", port=26500) # Create grpc channel
worker = ZeebeWorker(channel) # Create a zeebe worker
async def on_error(exception: Exception, job: Job):
"""
on_error will be called when the task fails
"""
print(exception)
await job.set_error_status(f"Failed to handle job {job}. Error: {str(exception)}")
@worker.task(task_type="example", exception_handler=on_error)
def example_task(input: str) -> dict:
return {"output": f"Hello world, {input}!"}
@worker.task(task_type="example2", exception_handler=on_error)
async def another_example_task(name: str) -> dict: # Tasks can also be async
return {"output": f"Hello world, {name} from async task!"}
loop = asyncio.get_running_loop()
loop.run_until_complete(worker.work()) # Now every time that a task with type `example` or `example2` is called, the corresponding function will be called
```
Stop a worker:
```python
await zeebe_worker.stop() # Stops worker after all running jobs have been completed
```
### Client
```python
from pyzeebe import ZeebeClient, create_insecure_channel
# Create a zeebe client
channel = create_insecure_channel(hostname="localhost", port=26500)
zeebe_client = ZeebeClient(channel)
# Run a Zeebe process instance
process_instance_key = await zeebe_client.run_process(bpmn_process_id="My zeebe process", variables={})
# Run a process and receive the result
process_instance_key, process_result = await zeebe_client.run_process_with_result(
bpmn_process_id="My zeebe process",
timeout=10000
)
# Deploy a BPMN process definition
await zeebe_client.deploy_process("process.bpmn")
# Cancel a running process
await zeebe_client.cancel_process_instance(process_instance_key=12345)
# Publish message
await zeebe_client.publish_message(name="message_name", correlation_key="some_id")
```
## Tests
Use the package manager [pip](https://pip.pypa.io/en/stable/) to install pyzeebe
`pytest tests/unit`
## Contributing
Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.
Please make sure to update tests as appropriate.
## Versioning
We use [SemVer](semver.org) for versioning. For the versions available, see the tags on this repository.
In order to bump the current version run:
```shell
$ bump2version <part>
```
where part is the part that will be bumped (major/minor/patch/rc).
This will bump the version in all relevant files as well as create a git commit.
## License
We use the MIT license, see [LICENSE.md](LICENSE.md) for details
Raw data
{
"_id": null,
"home_page": "https://github.com/Precise-Finance/pyzeebe",
"name": "pyzeebe-precise",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8,<4.0",
"maintainer_email": "",
"keywords": "zeebe,workflow,workflow-engine",
"author": "Jonatan Martens",
"author_email": "jonatanmartenstav@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/aa/7f/839e42cf0980f3c215f9c42eae4d195194fc78c4c5b3d979da06ef54cd54/pyzeebe_precise-3.0.5.tar.gz",
"platform": null,
"description": "[](https://github.com/camunda-community-hub/community)\n[](https://github.com/Camunda-Community-Hub/community/blob/main/extension-lifecycle.md#stable-)\n\n\n[](https://coveralls.io/github/JonatanMartens/pyzeebe?branch=master)\n\n\n\n\n\n\n\n[](https://pyzeebe.readthedocs.io/en/latest/?badge=stable)\n\n# Pyzeebe\n\npyzeebe is a python grpc client for Zeebe.\n\nZeebe version support:\n\n| Pyzeebe version | Tested Zeebe versions |\n| :-------------: | ---------------------- |\n| 3.x.x | 1.0.0 |\n| 2.x.x | 0.23, 0.24, 0.25, 0.26 |\n| 1.x.x | 0.23, 0.24 |\n\n## Getting Started\n\nTo install:\n\n`pip install pyzeebe`\n\nFor full documentation please visit: https://pyzeebe.readthedocs.io/en/stable/\n\n## Usage\n\n### Worker\n\nThe `ZeebeWorker` class gets jobs from the gateway and runs them.\n\n```python\nimport asyncio\n\nfrom pyzeebe import ZeebeWorker, Job, create_insecure_channel\n\n\nchannel = create_insecure_channel(hostname=\"localhost\", port=26500) # Create grpc channel\nworker = ZeebeWorker(channel) # Create a zeebe worker\n\n\nasync def on_error(exception: Exception, job: Job):\n \"\"\"\n on_error will be called when the task fails\n \"\"\"\n print(exception)\n await job.set_error_status(f\"Failed to handle job {job}. Error: {str(exception)}\")\n\n\n@worker.task(task_type=\"example\", exception_handler=on_error)\ndef example_task(input: str) -> dict:\n return {\"output\": f\"Hello world, {input}!\"}\n\n\n@worker.task(task_type=\"example2\", exception_handler=on_error)\nasync def another_example_task(name: str) -> dict: # Tasks can also be async\n return {\"output\": f\"Hello world, {name} from async task!\"}\n\nloop = asyncio.get_running_loop()\nloop.run_until_complete(worker.work()) # Now every time that a task with type `example` or `example2` is called, the corresponding function will be called\n```\n\nStop a worker:\n\n```python\nawait zeebe_worker.stop() # Stops worker after all running jobs have been completed\n```\n\n### Client\n\n```python\nfrom pyzeebe import ZeebeClient, create_insecure_channel\n\n# Create a zeebe client\nchannel = create_insecure_channel(hostname=\"localhost\", port=26500)\nzeebe_client = ZeebeClient(channel)\n\n# Run a Zeebe process instance\nprocess_instance_key = await zeebe_client.run_process(bpmn_process_id=\"My zeebe process\", variables={})\n\n# Run a process and receive the result\nprocess_instance_key, process_result = await zeebe_client.run_process_with_result(\n bpmn_process_id=\"My zeebe process\",\n timeout=10000\n)\n\n# Deploy a BPMN process definition\nawait zeebe_client.deploy_process(\"process.bpmn\")\n\n# Cancel a running process\nawait zeebe_client.cancel_process_instance(process_instance_key=12345)\n\n# Publish message\nawait zeebe_client.publish_message(name=\"message_name\", correlation_key=\"some_id\")\n\n```\n\n## Tests\n\nUse the package manager [pip](https://pip.pypa.io/en/stable/) to install pyzeebe\n\n`pytest tests/unit`\n\n## Contributing\n\nPull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.\n\nPlease make sure to update tests as appropriate.\n\n## Versioning\n\nWe use [SemVer](semver.org) for versioning. For the versions available, see the tags on this repository.\n\nIn order to bump the current version run:\n\n```shell\n$ bump2version <part>\n```\n\nwhere part is the part that will be bumped (major/minor/patch/rc).\n\nThis will bump the version in all relevant files as well as create a git commit.\n\n## License\n\nWe use the MIT license, see [LICENSE.md](LICENSE.md) for details\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Zeebe client api",
"version": "3.0.5",
"project_urls": {
"Documentation": "https://pyzeebe.readthedocs.io/en/latest",
"Homepage": "https://github.com/Precise-Finance/pyzeebe",
"Repository": "https://github.com/Precise-Finance/pyzeebe"
},
"split_keywords": [
"zeebe",
"workflow",
"workflow-engine"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "ea35760f9633dd49eb1fa164c6b544b7bef8621889f9450ca9dca4379417187c",
"md5": "c09c27d9363d80ded00afd9120857e1a",
"sha256": "3b234fe84553a00378cfbcc1d567ac649f189b846c1fd1cc9b29a4dd4806eabb"
},
"downloads": -1,
"filename": "pyzeebe_precise-3.0.5-py3-none-any.whl",
"has_sig": false,
"md5_digest": "c09c27d9363d80ded00afd9120857e1a",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8,<4.0",
"size": 31111,
"upload_time": "2024-01-15T10:19:37",
"upload_time_iso_8601": "2024-01-15T10:19:37.285736Z",
"url": "https://files.pythonhosted.org/packages/ea/35/760f9633dd49eb1fa164c6b544b7bef8621889f9450ca9dca4379417187c/pyzeebe_precise-3.0.5-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "aa7f839e42cf0980f3c215f9c42eae4d195194fc78c4c5b3d979da06ef54cd54",
"md5": "0945338c1e8070e1f19d0961bdb3edb2",
"sha256": "c79320f6e4ac15f585f5057c42bcf0648849866238f62ebf799063da13c37d9b"
},
"downloads": -1,
"filename": "pyzeebe_precise-3.0.5.tar.gz",
"has_sig": false,
"md5_digest": "0945338c1e8070e1f19d0961bdb3edb2",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8,<4.0",
"size": 20596,
"upload_time": "2024-01-15T10:19:39",
"upload_time_iso_8601": "2024-01-15T10:19:39.276499Z",
"url": "https://files.pythonhosted.org/packages/aa/7f/839e42cf0980f3c215f9c42eae4d195194fc78c4c5b3d979da06ef54cd54/pyzeebe_precise-3.0.5.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-01-15 10:19:39",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Precise-Finance",
"github_project": "pyzeebe",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "pyzeebe-precise"
}
JSON
