[](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": "https://github.com/camunda-community-hub/pyzeebe",
"name": "pyzeebe",
"maintainer": "Dmitriy",
"docs_url": null,
"requires_python": "<4.0,>=3.9",
"maintainer_email": "dimastbk@proton.me",
"keywords": "zeebe, workflow, workflow-engine",
"author": "Jonatan Martens",
"author_email": "jonatanmartenstav@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/11/f3/16d886b5bd2714f2d033b49a70849f37bb4f849022bd12bde0b8d6d7d099/pyzeebe-4.1.0.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": "MIT",
"summary": "Zeebe client api",
"version": "4.1.0",
"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": [
"zeebe",
" workflow",
" workflow-engine"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "d0559036507d2fb40705c31eae7158d9b5576305285d5d2dcab17448441bede3",
"md5": "49543cd73830c29b07352cc804b6e9c3",
"sha256": "888000184cb9878f9a392d8f1a67cfc2fb00a3782d7169527f8d975775d91223"
},
"downloads": -1,
"filename": "pyzeebe-4.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "49543cd73830c29b07352cc804b6e9c3",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.9",
"size": 39939,
"upload_time": "2024-11-11T07:16:22",
"upload_time_iso_8601": "2024-11-11T07:16:22.996694Z",
"url": "https://files.pythonhosted.org/packages/d0/55/9036507d2fb40705c31eae7158d9b5576305285d5d2dcab17448441bede3/pyzeebe-4.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "11f316d886b5bd2714f2d033b49a70849f37bb4f849022bd12bde0b8d6d7d099",
"md5": "e6c39028d6ab8f990fe243b0621cf639",
"sha256": "e09600ff50808eecc30ed86c48ed793a91f519adc13581990ea73ac28a784637"
},
"downloads": -1,
"filename": "pyzeebe-4.1.0.tar.gz",
"has_sig": false,
"md5_digest": "e6c39028d6ab8f990fe243b0621cf639",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.9",
"size": 26641,
"upload_time": "2024-11-11T07:16:24",
"upload_time_iso_8601": "2024-11-11T07:16:24.628708Z",
"url": "https://files.pythonhosted.org/packages/11/f3/16d886b5bd2714f2d033b49a70849f37bb4f849022bd12bde0b8d6d7d099/pyzeebe-4.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-11-11 07:16:24",
"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"
}
JSON
