Python local-first no-bloat task orchestration framework
Project description
woflo
Overview
woflo is a Python local-first no-bloat extensible task orchestration framework
Main goal is to abstract a lot of functionality related to Task orchestration and execution away.
Currently this includes:
- retries
- retry timeout
- parallelism
- logging
Examples
Intended usage is by utilizing a decorator @task
, consider a very simple example which would run 10 sleepy workers in parallel without blocking the main thread:
import time
from woflo import task
@task
def sleepy_worker():
time.sleep(5)
print('I am done')
for _ in range(10):
sleepy_task_run = sleepy_worker()
You can also include retries for tasks that might fail at times. Following would attempt to run the decorated function for 3 times in total with 5 second delay between attempts.
from woflo import task
@task(retries=2, retry_sleep_time=5)
def fetch_data_from_unstable_api():
...
Furthermore, you can also provide a runner within a @task
decorator. For example the SequentialTaskRun
if prefer your tasks to run sequentially and like to wait around a computer a lot. For example:
from woflo import task
from woflo.runners import SequentialTaskRun
@task(runner=SequentialTaskRun)
def sequential_sleepy_worker():
time.sleep(5)
print('I am done')
for _ in range(10):
sleepy_task_run = sequential_sleepy_worker()
Each TaskRun
should also expose a few methods that enable you to handle it:
.get_result()
to fetch the return value of the finished task.wait()
to block main thread till the task finishes (irrelevant forSequentialTaskRun
which will block until it finishes anyway).stop()
to stop the task while its running (irrelevant forSequentialTaskRun
which will block until it finishes anyway).is_running()
to check if the task is still running (irrelevant forSequentialTaskRun
which will block until it finishes anyway)
Let us define an example task:
import time
from woflo import task
@task
def quick_nap(duration):
time.sleep(duration)
if duration < 10
raise Exception("Ouch oof")
else:
return 'Well rested'
After you run it,
napping = quick_nap(10)
you can check on it to monitor it's state and receive results,
assert napping.is_running()
napping.wait()
assert napping.get_result() == "Well rested"
Extensibility
It is designed to be easily extended by developing a custom Task
runners. Library itself currently exposes two such runners, MultiprocessTaskRun
and SequentialTaskRun
.
Additionally woflo
makes available a BaseTaskRun
, an interface against which custom runners can be developed.
The defualt task runner is MultiprocessTaskRun
, which can run multiple tasks, or even multiple instances of the same task at the same time in parallel in separate Python process.
Roadmap
-
Setup GitHub Actions, SonarCloud monitoring and Codecov -
Make a PyPI Package - Decide on final API and create a version 1.x.x
- Implement a Dask runner
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.