A Python framework for stateful functional task dispatching and execution
Project description
FunTask
FunTask is a Python framework for stateful functional task dispatching and execution
- lightweight and easy
- scalable and distributed
- stateful
- task is just a
function(status, ...args)
- observable
from funtask import LocalFunTaskManager, StdLogger, TaskStatus
from typing import List
import asyncio
ft = LocalFunTaskManager(StdLogger())
# increase 10 worker(in different processes) with state: ['new_state']
workers = ft.increase_workers(lambda old_state: ['new_state'], 10)
def function_task(state: List[str], logger: StdLogger, *args):
state.append("change and visit state in func task")
logger.log(f"current state is {state}, args: '{' '.join(args)}'")
return "I'm task result"
# will dispatch function_task to worker0
task = workers[0].dispatch_fun_task(function_task, "i'm", "arguments")
# async wait task result
task_result = asyncio.run(task.get_result())
assert task_result == "I'm task result"
assert task.status is TaskStatus.SUCCESS
print("end.")
Table of Contents
Installation
not support for now
pip install funtask
License
funtask
is distributed under the terms of the MIT license.
Features
- local multiprocessing scheduler
- k8s + celeary task scheduler
- flet webUI
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
funtask-0.0.4.tar.gz
(21.1 MB
view hashes)