# jockey
Generic Python library for running asynchronous workers. Useful for building event handlers, web frameworks, and alike. You write a simple adapter, and jockey takes care of registering handlers, routing events, running jobs concurrently, priorities, capping the max number of workers, cancellation, waiting on exit, etc.
**Features:**
* Flexible, can be used as a web framework, as an event consumer, for background job processing, for parallel computing, and much more.
* Can run jobs in asyncio event loop, threads, or processes.
* 100% type safe.
* Reliable, battle-tested, and has 100% test coverage.
* Supports job priorities and all kinds of capping of the workers' number.
If you need an event-driven framework for distributed systems, take a look at [walnats](https://github.com/orsinium-labs/walnats/). It's a type-safe batteries-included Python library on top of [nats.py](https://github.com/nats-io/nats.py) designed to be safe, fast, and reliable. Jockey is based on walnats and the projects share lots of goals and design choices and even some API.
## Usage
Let's make a simple collection of functions for handling math operations.
Most of the classes jockey provides are generic and need to be parametrized. There are 3 type variables you need to define:
* `Payload` is the input argument for handlers. It's the request for web frameworks or the message for event handlers
* `Key` is the routing key which is used for selecting a handler. If is the URL path for web frameworks or routing key for event handlers.
* `Result` is the response of a handler. It's the response for web frameworks or None for event handlers.
For convenience, let's define them as type aliases:
```python
# each math operation accepts 2 integer numbers
Payload = tuple[int, int]
# each math operation is identified by a symbol (like "+" or "/")
Key = str
# each math operation returns a float
Result = float
```
The most important component to define is an Adapter. It's a collection of callbacks that tell how to convert a raw message into a routing key and a payload and how to handle successes and failures. In our case, we'll simply `print` from all callbacks:
```python
import asyncio
from dataclasses import dataclass
from typing import Iterator
import jockey
@dataclass
class Message(jockey.Adapter[Payload, Key, Result]):
left: int
op: Key
right: int
def get_keys(self) -> Iterator[Key]:
yield self.op
async def get_payload(self) -> Payload:
return (self.left, self.right)
async def on_success(self, result: Result) -> None:
print(f'SUCCESS: {self.left} {self.op} {self.right} = {result}')
async def on_failure(self, exc: Exception) -> None:
print(f'FAILURE: {self.left} {self.op} {self.right} caused {exc!r}')
async def on_cancel(self, exc: asyncio.CancelledError) -> None:
print(f'CANCELED: {self.left} {self.op} {self.right}')
```
Next, we make a registry of math operations:
```python
class Registry(jockey.Registry[Payload, Key, Result]):
pass
registry = Registry()
```
And in this registry, we can register all math operations ("handlers"):
```python
@registry.add('+')
def _add(payload: Payload) -> Result:
left, right = payload
return left + right
```
You can tell jockey to execute the task in a separate process (or thread):
```python
@registry.add('/', execute_in=jockey.ExecuteIn.PROCESS)
def _div(payload: Payload) -> Result:
left, right = payload
return left / right
```
Or make handlers async:
```python
@registry.add('-')
async def _sub(payload: Payload) -> Result:
left, right = payload
await asyncio.sleep(1)
return left / right
```
And the last thing, we make an executor and schedule messages:
```python
async def main() -> None:
async with jockey.Executor(registry).run() as executor:
messages = [
Message(3, '-', 2),
Message(4, '+', 5),
Message(3, '/', 2),
Message(3, '/', 0),
Message(3, '+', 0),
]
for msg in messages:
await executor.execute(msg)
if __name__ == '__main__':
asyncio.run(main())
```
That's it! The output should look like this:
```yaml
SUCCESS: 4 + 5 = 9
SUCCESS: 3 + 0 = 3
SUCCESS: 3 / 2 = 1.5
FAILURE: 3 / 0 caused ZeroDivisionError('division by zero')
SUCCESS: 3 - 2 = 1.5
```
Notice that we send `Message(3, '-', 2)` first but because of `await asyncio.sleep(1)` it arrived last. It show 2 important things:
1. `executor.execute` runs all messages concurrently. So, while the first message is blocked, the rest can be processed. You canspecify for how long when you want the executor to return with the `wait_for` argument.
1. When leaving the context, the executor will block and wait for all messages to finish. Similarly, if the executor is cancelled, it will make sure to cancel all running handlers (and execute `Adapter.on_cancel`).
## More examples
The [examples](./examples/) directory has some examples that you can run yourself and see the projecti n action:
* [math.py](./examples/math.py) is the code from the Usage section above.
* [rabbitmq_consumer.py](./examples/rabbitmq_consumer.py) shows how to consume and handle messages from RabbitMQ using jockey and [aio-pika](https://github.com/mosquito/aio-pika).
* [web_framework.py](./examples/web_framework.py) shows how to write a simple web framework using jockey and [uvicorn](https://www.uvicorn.org/).
## Advanced usage
The library can do surprisingly many things but nobody would read a big wordy documentation about what and why. Instead, we provide a few examples (see above) and extensive docstrings for every public method. Hence the best way to get started is to take one of the examples, adjust it for your needs, and then have a look at docstrings of the stuff you use to see what you can and should configure.
## QnA
1. **Is it maintained?** The project is pretty much feature-complete, so there is nothing for me to commit and release daily. However, I accept contributions (see below).
1. **What if I found a bug?** Fork the project, fix the bug, write some tests, and open a Pull Request. I usually merge and release any contributions within a day.
1. **Does it have retries?** Web frameworks don't do retries and event consumers handle retries on the message broker side, so for most use cases having retries in the jockey itself is redundant. You can always implement your own retry logic by sending a new message from `Adapter.on_failure`.
Raw data
{
"_id": null,
"home_page": null,
"name": "jockey",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "microservices,event-driven,distributed,web,framewrok,web framework,asyncio,kafka,nats,rabbitmq,arq,celery,dramatiq,huey,rq,walnats",
"author": null,
"author_email": "Gram <git@orsinium.dev>",
"download_url": "https://files.pythonhosted.org/packages/5c/01/502b749a8ee8d81864719f50bd3c832285da20347647c29e5685e274e86b/jockey-1.0.0.tar.gz",
"platform": null,
"description": "# jockey\n\nGeneric Python library for running asynchronous workers. Useful for building event handlers, web frameworks, and alike. You write a simple adapter, and jockey takes care of registering handlers, routing events, running jobs concurrently, priorities, capping the max number of workers, cancellation, waiting on exit, etc.\n\n**Features:**\n\n* Flexible, can be used as a web framework, as an event consumer, for background job processing, for parallel computing, and much more.\n* Can run jobs in asyncio event loop, threads, or processes.\n* 100% type safe.\n* Reliable, battle-tested, and has 100% test coverage.\n* Supports job priorities and all kinds of capping of the workers' number.\n\nIf you need an event-driven framework for distributed systems, take a look at [walnats](https://github.com/orsinium-labs/walnats/). It's a type-safe batteries-included Python library on top of [nats.py](https://github.com/nats-io/nats.py) designed to be safe, fast, and reliable. Jockey is based on walnats and the projects share lots of goals and design choices and even some API.\n\n## Usage\n\nLet's make a simple collection of functions for handling math operations.\n\nMost of the classes jockey provides are generic and need to be parametrized. There are 3 type variables you need to define:\n\n* `Payload` is the input argument for handlers. It's the request for web frameworks or the message for event handlers\n* `Key` is the routing key which is used for selecting a handler. If is the URL path for web frameworks or routing key for event handlers.\n* `Result` is the response of a handler. It's the response for web frameworks or None for event handlers.\n\nFor convenience, let's define them as type aliases:\n\n```python\n# each math operation accepts 2 integer numbers\nPayload = tuple[int, int]\n# each math operation is identified by a symbol (like \"+\" or \"/\")\nKey = str\n# each math operation returns a float\nResult = float\n```\n\nThe most important component to define is an Adapter. It's a collection of callbacks that tell how to convert a raw message into a routing key and a payload and how to handle successes and failures. In our case, we'll simply `print` from all callbacks:\n\n```python\nimport asyncio\nfrom dataclasses import dataclass\nfrom typing import Iterator\nimport jockey\n\n@dataclass\nclass Message(jockey.Adapter[Payload, Key, Result]):\n left: int\n op: Key\n right: int\n\n def get_keys(self) -> Iterator[Key]:\n yield self.op\n\n async def get_payload(self) -> Payload:\n return (self.left, self.right)\n\n async def on_success(self, result: Result) -> None:\n print(f'SUCCESS: {self.left} {self.op} {self.right} = {result}')\n\n async def on_failure(self, exc: Exception) -> None:\n print(f'FAILURE: {self.left} {self.op} {self.right} caused {exc!r}')\n\n async def on_cancel(self, exc: asyncio.CancelledError) -> None:\n print(f'CANCELED: {self.left} {self.op} {self.right}')\n```\n\nNext, we make a registry of math operations:\n\n```python\nclass Registry(jockey.Registry[Payload, Key, Result]):\n pass\n\nregistry = Registry()\n```\n\nAnd in this registry, we can register all math operations (\"handlers\"):\n\n```python\n@registry.add('+')\ndef _add(payload: Payload) -> Result:\n left, right = payload\n return left + right\n```\n\nYou can tell jockey to execute the task in a separate process (or thread):\n\n```python\n@registry.add('/', execute_in=jockey.ExecuteIn.PROCESS)\ndef _div(payload: Payload) -> Result:\n left, right = payload\n return left / right\n```\n\nOr make handlers async:\n\n```python\n@registry.add('-')\nasync def _sub(payload: Payload) -> Result:\n left, right = payload\n await asyncio.sleep(1)\n return left / right\n```\n\nAnd the last thing, we make an executor and schedule messages:\n\n```python\nasync def main() -> None:\n async with jockey.Executor(registry).run() as executor:\n messages = [\n Message(3, '-', 2),\n Message(4, '+', 5),\n Message(3, '/', 2),\n Message(3, '/', 0),\n Message(3, '+', 0),\n ]\n for msg in messages:\n await executor.execute(msg)\n\nif __name__ == '__main__':\n asyncio.run(main())\n```\n\nThat's it! The output should look like this:\n\n```yaml\nSUCCESS: 4 + 5 = 9\nSUCCESS: 3 + 0 = 3\nSUCCESS: 3 / 2 = 1.5\nFAILURE: 3 / 0 caused ZeroDivisionError('division by zero')\nSUCCESS: 3 - 2 = 1.5\n```\n\nNotice that we send `Message(3, '-', 2)` first but because of `await asyncio.sleep(1)` it arrived last. It show 2 important things:\n\n1. `executor.execute` runs all messages concurrently. So, while the first message is blocked, the rest can be processed. You canspecify for how long when you want the executor to return with the `wait_for` argument.\n1. When leaving the context, the executor will block and wait for all messages to finish. Similarly, if the executor is cancelled, it will make sure to cancel all running handlers (and execute `Adapter.on_cancel`).\n\n## More examples\n\nThe [examples](./examples/) directory has some examples that you can run yourself and see the projecti n action:\n\n* [math.py](./examples/math.py) is the code from the Usage section above.\n* [rabbitmq_consumer.py](./examples/rabbitmq_consumer.py) shows how to consume and handle messages from RabbitMQ using jockey and [aio-pika](https://github.com/mosquito/aio-pika).\n* [web_framework.py](./examples/web_framework.py) shows how to write a simple web framework using jockey and [uvicorn](https://www.uvicorn.org/).\n\n## Advanced usage\n\nThe library can do surprisingly many things but nobody would read a big wordy documentation about what and why. Instead, we provide a few examples (see above) and extensive docstrings for every public method. Hence the best way to get started is to take one of the examples, adjust it for your needs, and then have a look at docstrings of the stuff you use to see what you can and should configure.\n\n## QnA\n\n1. **Is it maintained?** The project is pretty much feature-complete, so there is nothing for me to commit and release daily. However, I accept contributions (see below).\n1. **What if I found a bug?** Fork the project, fix the bug, write some tests, and open a Pull Request. I usually merge and release any contributions within a day.\n1. **Does it have retries?** Web frameworks don't do retries and event consumers handle retries on the message broker side, so for most use cases having retries in the jockey itself is redundant. You can always implement your own retry logic by sending a new message from `Adapter.on_failure`.\n",
"bugtrack_url": null,
"license": null,
"summary": "Generic and type safe runner for asynchronous workers.",
"version": "1.0.0",
"project_urls": {
"Source": "https://github.com/orsinium-labs/jockey"
},
"split_keywords": [
"microservices",
"event-driven",
"distributed",
"web",
"framewrok",
"web framework",
"asyncio",
"kafka",
"nats",
"rabbitmq",
"arq",
"celery",
"dramatiq",
"huey",
"rq",
"walnats"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "9b614e262baf6d2dc9484517b7ac364a030c654ef80a0374b954b5053abea02f",
"md5": "61606e7ae67e9bdd5d400f621ad5dc80",
"sha256": "721fc5e7214ca381724feade034f93998c08568cf0ea6cc3f8521bb3f41655aa"
},
"downloads": -1,
"filename": "jockey-1.0.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "61606e7ae67e9bdd5d400f621ad5dc80",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 13961,
"upload_time": "2023-11-03T09:34:27",
"upload_time_iso_8601": "2023-11-03T09:34:27.836343Z",
"url": "https://files.pythonhosted.org/packages/9b/61/4e262baf6d2dc9484517b7ac364a030c654ef80a0374b954b5053abea02f/jockey-1.0.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "5c01502b749a8ee8d81864719f50bd3c832285da20347647c29e5685e274e86b",
"md5": "23c14a586722e77549757ee81c02a6a8",
"sha256": "3c7de864d7cb91a3b19ad4556e229819b8f8bb18afb9f72f190c2e4ce2287e27"
},
"downloads": -1,
"filename": "jockey-1.0.0.tar.gz",
"has_sig": false,
"md5_digest": "23c14a586722e77549757ee81c02a6a8",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 19377,
"upload_time": "2023-11-03T09:34:30",
"upload_time_iso_8601": "2023-11-03T09:34:30.150372Z",
"url": "https://files.pythonhosted.org/packages/5c/01/502b749a8ee8d81864719f50bd3c832285da20347647c29e5685e274e86b/jockey-1.0.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-11-03 09:34:30",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "orsinium-labs",
"github_project": "jockey",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "jockey"
}
JSON
