# Dispatch Highlevel Interface
[](https://github.com/frequenz-floss/frequenz-dispatch-python/actions/workflows/ci.yaml)
[](https://pypi.org/project/frequenz-dispatch/)
[](https://frequenz-floss.github.io/frequenz-dispatch-python/)
## Introduction
A highlevel interface for the dispatch API.
See [the documentation](https://frequenz-floss.github.io/frequenz-dispatch-python/v0.1/reference/frequenz/dispatch) for more information.
## Usage
The [`Dispatcher` class](https://frequenz-floss.github.io/frequenz-dispatch-python/v0.1/reference/frequenz/dispatch/#frequenz.dispatch.Dispatcher), the main entry point for the API, provides two channels:
* [Lifecycle events](https://frequenz-floss.github.io/frequenz-dispatch-python/v0.1/reference/frequenz/dispatch/#frequenz.dispatch.Dispatcher.lifecycle_events): A channel that sends a message whenever a [Dispatch][frequenz.dispatch.Dispatch] is created, updated or deleted.
* [Running status change](https://frequenz-floss.github.io/frequenz-dispatch-python/v0.1/reference/frequenz/dispatch/#frequenz.dispatch.Dispatcher.running_status_change): Sends a dispatch message whenever a dispatch is ready to be executed according to the schedule or the running status of the dispatch changed in a way that could potentially require the actor to start, stop or reconfigure itself.
### Example using the running status change channel
```python
import os
from unittest.mock import MagicMock
from datetime import timedelta
from frequenz.dispatch import Dispatcher, DispatchInfo, MergeByType
async def create_actor(dispatch: DispatchInfo, receiver: Receiver[DispatchInfo]) -> Actor:
return MagicMock(dispatch=dispatch, receiver=receiver)
async def run():
url = os.getenv("DISPATCH_API_URL", "grpc://dispatch.url.goes.here.example.com")
key = os.getenv("DISPATCH_API_KEY", "some-key")
microgrid_id = 1
async with Dispatcher(
microgrid_id=microgrid_id,
server_url=url,
key=key,
) as dispatcher:
await dispatcher.start_managing(
dispatch_type="EXAMPLE_TYPE",
actor_factory=create_actor,
merge_strategy=MergeByType(),
retry_interval=timedelta(seconds=10)
)
await dispatcher
```
## Supported Platforms
The following platforms are officially supported (tested):
- **Python:** 3.11
- **Operating System:** Ubuntu Linux 20.04
- **Architectures:** amd64, arm64
## Contributing
If you want to know how to build this project and contribute to it, please
check out the [Contributing Guide](CONTRIBUTING.md).
Raw data
{
"_id": null,
"home_page": null,
"name": "frequenz-dispatch",
"maintainer": null,
"docs_url": null,
"requires_python": "<4,>=3.11",
"maintainer_email": null,
"keywords": "frequenz, python, actor, frequenz-dispatch, dispatch, highlevel, api",
"author": null,
"author_email": "Frequenz Energy-as-a-Service GmbH <floss@frequenz.com>",
"download_url": "https://files.pythonhosted.org/packages/bc/03/7951045e7ee4ec692fd4587c25a6537fa9e06778479ab9dfc39ba7ed762a/frequenz_dispatch-0.10.2.tar.gz",
"platform": null,
"description": "# Dispatch Highlevel Interface\n\n[](https://github.com/frequenz-floss/frequenz-dispatch-python/actions/workflows/ci.yaml)\n[](https://pypi.org/project/frequenz-dispatch/)\n[](https://frequenz-floss.github.io/frequenz-dispatch-python/)\n\n## Introduction\n\nA highlevel interface for the dispatch API.\n\nSee [the documentation](https://frequenz-floss.github.io/frequenz-dispatch-python/v0.1/reference/frequenz/dispatch) for more information.\n\n## Usage\n\nThe [`Dispatcher` class](https://frequenz-floss.github.io/frequenz-dispatch-python/v0.1/reference/frequenz/dispatch/#frequenz.dispatch.Dispatcher), the main entry point for the API, provides two channels:\n\n* [Lifecycle events](https://frequenz-floss.github.io/frequenz-dispatch-python/v0.1/reference/frequenz/dispatch/#frequenz.dispatch.Dispatcher.lifecycle_events): A channel that sends a message whenever a [Dispatch][frequenz.dispatch.Dispatch] is created, updated or deleted.\n* [Running status change](https://frequenz-floss.github.io/frequenz-dispatch-python/v0.1/reference/frequenz/dispatch/#frequenz.dispatch.Dispatcher.running_status_change): Sends a dispatch message whenever a dispatch is ready to be executed according to the schedule or the running status of the dispatch changed in a way that could potentially require the actor to start, stop or reconfigure itself.\n\n### Example using the running status change channel\n\n```python\nimport os\nfrom unittest.mock import MagicMock\nfrom datetime import timedelta\n\nfrom frequenz.dispatch import Dispatcher, DispatchInfo, MergeByType\n\nasync def create_actor(dispatch: DispatchInfo, receiver: Receiver[DispatchInfo]) -> Actor:\n return MagicMock(dispatch=dispatch, receiver=receiver)\n\nasync def run():\n url = os.getenv(\"DISPATCH_API_URL\", \"grpc://dispatch.url.goes.here.example.com\")\n key = os.getenv(\"DISPATCH_API_KEY\", \"some-key\")\n\n microgrid_id = 1\n\n async with Dispatcher(\n microgrid_id=microgrid_id,\n server_url=url,\n key=key,\n ) as dispatcher:\n await dispatcher.start_managing(\n dispatch_type=\"EXAMPLE_TYPE\",\n actor_factory=create_actor,\n merge_strategy=MergeByType(),\n retry_interval=timedelta(seconds=10)\n )\n\n await dispatcher\n```\n\n## Supported Platforms\n\nThe following platforms are officially supported (tested):\n\n- **Python:** 3.11\n- **Operating System:** Ubuntu Linux 20.04\n- **Architectures:** amd64, arm64\n\n## Contributing\n\nIf you want to know how to build this project and contribute to it, please\ncheck out the [Contributing Guide](CONTRIBUTING.md).\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A highlevel interface for the dispatch API",
"version": "0.10.2",
"project_urls": {
"Changelog": "https://github.com/frequenz-floss/frequenz-dispatch-python/releases",
"Documentation": "https://frequenz-floss.github.io/frequenz-dispatch-python/",
"Issues": "https://github.com/frequenz-floss/frequenz-dispatch-python/issues",
"Repository": "https://github.com/frequenz-floss/frequenz-dispatch-python",
"Support": "https://github.com/frequenz-floss/frequenz-dispatch-python/discussions/categories/support"
},
"split_keywords": [
"frequenz",
" python",
" actor",
" frequenz-dispatch",
" dispatch",
" highlevel",
" api"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "2b8c75c8a170fc4854bb6c6e4a856da5e02664f652a98c954e981a1df424619d",
"md5": "e97d08aef17b237034da16943cadd5d0",
"sha256": "fd05a1906c796224c58d921d3c0a627b7c39fb475e24370cb0f13211b1eea0e4"
},
"downloads": -1,
"filename": "frequenz_dispatch-0.10.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "e97d08aef17b237034da16943cadd5d0",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4,>=3.11",
"size": 21476,
"upload_time": "2025-07-29T08:18:17",
"upload_time_iso_8601": "2025-07-29T08:18:17.063186Z",
"url": "https://files.pythonhosted.org/packages/2b/8c/75c8a170fc4854bb6c6e4a856da5e02664f652a98c954e981a1df424619d/frequenz_dispatch-0.10.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "bc037951045e7ee4ec692fd4587c25a6537fa9e06778479ab9dfc39ba7ed762a",
"md5": "f49d3648ce0238b565be6def9efd311e",
"sha256": "02d1ae469f0dc5e9bd92f457b1135c958737c7f4a40b7021c35b50c2e2eb8f01"
},
"downloads": -1,
"filename": "frequenz_dispatch-0.10.2.tar.gz",
"has_sig": false,
"md5_digest": "f49d3648ce0238b565be6def9efd311e",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4,>=3.11",
"size": 21700,
"upload_time": "2025-07-29T08:18:18",
"upload_time_iso_8601": "2025-07-29T08:18:18.372996Z",
"url": "https://files.pythonhosted.org/packages/bc/03/7951045e7ee4ec692fd4587c25a6537fa9e06778479ab9dfc39ba7ed762a/frequenz_dispatch-0.10.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-29 08:18:18",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "frequenz-floss",
"github_project": "frequenz-dispatch-python",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "frequenz-dispatch"
}
JSON
