| Name | pyzeebe JSON |
| Version |
4.6.1
JSON |
| download |
| home_page | None |
| Summary | Zeebe client api |
| upload_time | 2025-08-05 05:23:22 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.9 |
| license | None |
| keywords |
workflow
workflow-engine
zeebe
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
[](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)
# Pyzeebe
pyzeebe is a python grpc client for Zeebe.
Zeebe version support:
| Pyzeebe version | Tested Zeebe versions |
| :-------------: | ---------------------- |
| 4.x.x | 8.2, 8.3, 8.4, 8.5 |
| 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://camunda-community-hub.github.io/pyzeebe/
## Usage
### Worker
The `ZeebeWorker` class gets jobs from the gateway and runs them.
```python
import asyncio
from pyzeebe import ZeebeWorker, Job, JobController, create_insecure_channel
channel = create_insecure_channel(grpc_address="localhost:26500") # Create grpc channel
worker = ZeebeWorker(channel) # Create a zeebe worker
async def on_error(exception: Exception, job: Job, job_controller: JobController):
"""
on_error will be called when the task fails
"""
print(exception)
await job_controller.set_error_status(job, 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(grpc_address="localhost: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_resource("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.
## License
We use the MIT license, see [LICENSE.md](LICENSE.md) for details
Raw data
{
"_id": null,
"home_page": null,
"name": "pyzeebe",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": "Dmitriy <dimastbk@proton.me>",
"keywords": "workflow, workflow-engine, zeebe",
"author": null,
"author_email": "Jonatan Martens <jonatanmartenstav@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/53/f1/8274c9109d1c5288662872e78026d76d491eac9dc1b75836386aa6d538b9/pyzeebe-4.6.1.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\n# Pyzeebe\n\npyzeebe is a python grpc client for Zeebe.\n\nZeebe version support:\n\n| Pyzeebe version | Tested Zeebe versions |\n| :-------------: | ---------------------- |\n| 4.x.x | 8.2, 8.3, 8.4, 8.5 |\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://camunda-community-hub.github.io/pyzeebe/\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, JobController, create_insecure_channel\n\n\nchannel = create_insecure_channel(grpc_address=\"localhost:26500\") # Create grpc channel\nworker = ZeebeWorker(channel) # Create a zeebe worker\n\n\nasync def on_error(exception: Exception, job: Job, job_controller: JobController):\n \"\"\"\n on_error will be called when the task fails\n \"\"\"\n print(exception)\n await job_controller.set_error_status(job, 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(grpc_address=\"localhost: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_resource(\"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\n## License\n\nWe use the MIT license, see [LICENSE.md](LICENSE.md) for details\n",
"bugtrack_url": null,
"license": null,
"summary": "Zeebe client api",
"version": "4.6.1",
"project_urls": {
"documentation": "https://camunda-community-hub.github.io/pyzeebe/",
"homepage": "https://github.com/camunda-community-hub/pyzeebe",
"repository": "https://github.com/camunda-community-hub/pyzeebe"
},
"split_keywords": [
"workflow",
" workflow-engine",
" zeebe"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "fada7e823530a56a113aca9cf97baa872ced3a849bb0f0867bdc75da628fd005",
"md5": "1dd36dff864a3f547a6349a6de6b9773",
"sha256": "fe3479c67ac83dabb2a197a76a500209907c274d93ef837c160606e72c33e28f"
},
"downloads": -1,
"filename": "pyzeebe-4.6.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "1dd36dff864a3f547a6349a6de6b9773",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 70853,
"upload_time": "2025-08-05T05:23:21",
"upload_time_iso_8601": "2025-08-05T05:23:21.154215Z",
"url": "https://files.pythonhosted.org/packages/fa/da/7e823530a56a113aca9cf97baa872ced3a849bb0f0867bdc75da628fd005/pyzeebe-4.6.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "53f18274c9109d1c5288662872e78026d76d491eac9dc1b75836386aa6d538b9",
"md5": "673fefbf91918cdeb9db3a5ba9d95d67",
"sha256": "83f89b9e04cf60a1ec4c4baa3fe807a00bee1c40d4dd1c0fd750355039b05358"
},
"downloads": -1,
"filename": "pyzeebe-4.6.1.tar.gz",
"has_sig": false,
"md5_digest": "673fefbf91918cdeb9db3a5ba9d95d67",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 149837,
"upload_time": "2025-08-05T05:23:22",
"upload_time_iso_8601": "2025-08-05T05:23:22.585014Z",
"url": "https://files.pythonhosted.org/packages/53/f1/8274c9109d1c5288662872e78026d76d491eac9dc1b75836386aa6d538b9/pyzeebe-4.6.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-05 05:23:22",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "camunda-community-hub",
"github_project": "pyzeebe",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "pyzeebe"
}
