A simple Python module to help distribute you tasks across multiple brokers as microservices
Project description
Classwork
This is a simple Python module to help distribute you tasks across multiple brokers as microservices.
Classwork uses NATS to manage communication between job schedulers and workers. Nats does most of the heavy lifting, allowing us to keep the rest of the code simple and concise.
What is in name?
So why ClassWork? Well, this is simply because workers are simply "python classes" (The Class) whose "methods" become individual workers (The Students).
Get started
First let us create a simple worker
import asyncio
from classwork import ClassWork
class WorkerClass:
# Your worker...
# Notice that we expect workers to be async
async def add(self, a, b):
# the work!
resp = a + b
# simulate a long process
await asyncio.sleep(3)
# return results
return resp
# this is our nats server url
nats_url = "nats://127.0.0.1:4222"
# initialize ClassWork
class_work = ClassWork(nats_url=nats_url)
# init worker class
worker = WorkerClass()
# Ok, let us register things with asyncio
# notice the method 'class_work.register' is async!
asyncio.run(class_work.register(name="my-worker", worker_class=worker))
This is all the code we need to set up the worker. It is important to note the following:
- Worker Class methods should be asynchronous
- class_work.register is also async
- You will need NATS running with JetStream enabled!
The Job Scheduler
Now we need to create the job scheduler. Below is the code we need
import asyncio
import numpy as np
from pprint import pprint
from classwork import ClassWork
# init ClassWork
nats_url = "nats://127.0.0.1:4222"
class_work = ClassWork(nats_url=nats_url)
# Our callback function
# This is where complete work gets reported
def report_callback(report_card):
print("We have a report!")
pprint(report_card)
# We need this function to create our job schedules
async def schedules():
# Assign a job
await class_work.assign(
# the task name
task="my_worker.add",
# arguments
args=[1, 2],
# the callback to report our results
report_callback=report_callback,
)
# Ok, let us create the schedules now
asyncio.run(schedules())
This code will create a task into NATS and the job workers (attentive students 😂) already listening will pick the task and run it, then publish their reports which is routed via NATS back to the scheduler (teacher?).
Take note of the following:
class_work.assign
must be run in async mode. So we have wrapped it in an async method. You can also useasyncio.run
directly.- Naming your task is very important. This naming convection is unashamedly borrowed from moleculer. In this case, your task is "my_worker.add". This will route to any worker class registered with the name "my_worker" and method "add".
- Because all this traffic is routed via NATS, your arguments must be JSON serializable. Even though we use typ to handle edge cases like
sets
, beware that there are limits to what you can pass in your arguments report_callback
is called with a 'report' of your task. A report card 😊 will look like the one below:
We have a report!
{'task': 'my_worker.add',
'duration': {'latency': {'request': '4 ms and 356 µs',
'response': '5 s and 4 ms'},
'my_worker.add': '3 s and 1 ms'},
'req_id': '8mYjJjM0kb5',
'response': 3}
Report Explanation
- duration: is a high precision (down to yoctoseconds) report of the time taken.
- latency: shows the time taken to route your "task request" to the worker and "task response" back to the scheduler.
- req_id: is a unique id assigned to each job
- response: is the actual value returned by the worker
Try it
We have sample code in scheduler.py and worker.py
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for classwork-0.0.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 1d6b514b6581e8349c6f9795ff09dd52e17e4937f7e0d6e3e4165ed3c5843468 |
|
MD5 | 182fc4d89f5d7ab5d2c3f438247e01e9 |
|
BLAKE2b-256 | 0416411503b6a988c1c5bff98e0e5de8617b0147d2fe8ae008bfb75ea69587b1 |