# Model tasks
An opinionated and lightweight workflow management system and task graph. Born from the desire to have
a reusable code skeleton for geoprocessing and data pipelines projects this package offers:
- Write cleaner code code in separate tasks
- Formulate task dependencies and automatically resolve those as a direct acyclical graph (DAG)
- Display your models
- Parallelize concurrent tasks
- Task configuration
- Caching and invalidation of intermediate task results based on configuration
- Easily generate a task graph documentation
## Quickstart
A model consists of individual tasks which formulate their dependence on other tasks. Together they build one or more acyclical directed graphs, which do not allow loops or task repetition. Tasks are implement as subclasses of the ModelTask class. They can be either all defined within one file or within a folder of Python modules. The latter is more practical if the model grows and consists of many tasks.
### Create a one-file model
**model.py**
```
from modeltasks import Model, ModelTask
my_model = Model(title='My Model', model_tasks=__file__)
class TaskA(ModelTask):
def run (self, logger, workspace):
logger.info(f'Running an A task in {workspace}')
class TaskB(ModelTask):
def run (self, logger, workspace):
logger.info(f'Running a B task in {workspace}')
```
### Create a model with task modules
**Prepare project structure**
```
touch model.py
mkdir task_modules
touch task_modules/a_tasks.py
touch task_modules/b_tasks.py
```
**model.py**
```
from modeltasks import Model
my_model = Model(title='My Model', model_tasks='task_modules')
```
**a_tasks.py**
```
from modeltasks import ModelTask
class TaskA(ModelTask):
def run (self, logger, workspace):
logger.info(f'Running an A task in {workspace}')
```
**b_tasks.py**
```
from modeltasks import ModelTask
class TaskB(ModelTask):
def run (self, logger, workspace):
logger.info(f'Running a B task in {workspace}')
```
The above code creates a model with two simple tasks. But not a task graph yet because both tasks have not defined:
- inputs (dependencies)
- outputs (results)
- configuration
Putting tasks into a workflow graph is achieved by defining task inputs and outputs. Let's look at the two example tasks we just created and assume that `TaskB` requires `TaskA` to run first and then use its output.
**a_tasks.py (With output)**
```
from modeltasks import ModelTask
from modeltasks.data import VariableOutput
class TaskA(ModelTask):
a_output: VariableOutput
def run (self, logger, workspace):
logger.info(f'Running an A task in {workspace}')
self.a_output = 'First I ran task A.'
```
**b_tasks.py (With dependency and output)**
```
from modeltasks import ModelTask
from modeltasks.data import VariableInput, VariableOutput
class TaskB(ModelTask):
a_input: VariableInput = 'a_tasks.TaskA'
b_output: VariableOutput
def run (self, logger, workspace):
logger.info(f'Running an B task in {workspace}')
self.b_output = f'{self.a_input} Then I ran task B.'
```
### Run a model
To run a model, we need to specify an entry task. This is the task that will be run at the end after all of its required task dependencies have been resolved and their output gathered.
```
python3 model.py --run --task b_tasks.TaskB
```
### Model visualization
Sometimes it is helpful to see a visual representation of all the task interdependencies. To render such a visual task graph call your model with:
```
python3 model.py --graph --output=mermaid.md (Mermaid file)
python3 model.py --graph --output=graph.png (Image file)
```
## Documentation
To learn more about supported input and output types, dependency definition, task schedulers, result caching, etc. head over to the [package documentation](https://gitlab.com/geotom/modeltasks/-/wikis/Modeltasks)
## Contribution
Please leave feedback, questions, suggestions on the [project's issue tracker](https://gitlab.com/geotom/modeltasks/-/issues).
Raw data
{
"_id": null,
"home_page": "https://gitlab.com/geotom/modeltasks",
"name": "modeltasks",
"maintainer": null,
"docs_url": null,
"requires_python": null,
"maintainer_email": null,
"keywords": "Tasks, Model, DAG, Graph, Model, Processing, Workflow management",
"author": "Thomas Wanderer",
"author_email": "contact-project+geotom-modeltasks-support@incoming.gitlab.com",
"download_url": null,
"platform": null,
"description": "# Model tasks\n\nAn opinionated and lightweight workflow management system and task graph. Born from the desire to have\na reusable code skeleton for geoprocessing and data pipelines projects this package offers:\n\n- Write cleaner code code in separate tasks\n- Formulate task dependencies and automatically resolve those as a direct acyclical graph (DAG)\n- Display your models\n- Parallelize concurrent tasks\n- Task configuration\n- Caching and invalidation of intermediate task results based on configuration\n- Easily generate a task graph documentation\n\n\n## Quickstart\n\nA model consists of individual tasks which formulate their dependence on other tasks. Together they build one or more acyclical directed graphs, which do not allow loops or task repetition. Tasks are implement as subclasses of the ModelTask class. They can be either all defined within one file or within a folder of Python modules. The latter is more practical if the model grows and consists of many tasks.\n\n### Create a one-file model\n\n**model.py**\n\n```\nfrom modeltasks import Model, ModelTask\n\n\nmy_model = Model(title='My Model', model_tasks=__file__)\n\n\nclass TaskA(ModelTask):\n def run (self, logger, workspace):\n logger.info(f'Running an A task in {workspace}')\n \n \nclass TaskB(ModelTask):\n def run (self, logger, workspace):\n logger.info(f'Running a B task in {workspace}')\n```\n\n### Create a model with task modules\n\n**Prepare project structure**\n\n```\ntouch model.py\nmkdir task_modules\ntouch task_modules/a_tasks.py\ntouch task_modules/b_tasks.py\n```\n\n**model.py**\n\n```\nfrom modeltasks import Model\n\n\nmy_model = Model(title='My Model', model_tasks='task_modules')\n```\n\n**a_tasks.py**\n\n```\nfrom modeltasks import ModelTask\n\n\nclass TaskA(ModelTask):\n def run (self, logger, workspace):\n logger.info(f'Running an A task in {workspace}')\n```\n\n**b_tasks.py**\n\n```\nfrom modeltasks import ModelTask\n\n\nclass TaskB(ModelTask):\n def run (self, logger, workspace):\n logger.info(f'Running a B task in {workspace}')\n```\n\nThe above code creates a model with two simple tasks. But not a task graph yet because both tasks have not defined:\n- inputs (dependencies)\n- outputs (results)\n- configuration\n\nPutting tasks into a workflow graph is achieved by defining task inputs and outputs. Let's look at the two example tasks we just created and assume that `TaskB` requires `TaskA` to run first and then use its output.\n\n**a_tasks.py (With output)**\n\n```\nfrom modeltasks import ModelTask\nfrom modeltasks.data import VariableOutput\n\n\nclass TaskA(ModelTask):\n\n a_output: VariableOutput\n\n def run (self, logger, workspace):\n logger.info(f'Running an A task in {workspace}')\n self.a_output = 'First I ran task A.'\n```\n\n**b_tasks.py (With dependency and output)**\n\n```\nfrom modeltasks import ModelTask\nfrom modeltasks.data import VariableInput, VariableOutput\n\n\nclass TaskB(ModelTask):\n\n a_input: VariableInput = 'a_tasks.TaskA'\n b_output: VariableOutput\n\n def run (self, logger, workspace):\n logger.info(f'Running an B task in {workspace}')\n self.b_output = f'{self.a_input} Then I ran task B.'\n```\n\n### Run a model\n\nTo run a model, we need to specify an entry task. This is the task that will be run at the end after all of its required task dependencies have been resolved and their output gathered.\n\n```\npython3 model.py --run --task b_tasks.TaskB\n```\n\n### Model visualization\n\nSometimes it is helpful to see a visual representation of all the task interdependencies. To render such a visual task graph call your model with:\n\n```\npython3 model.py --graph --output=mermaid.md (Mermaid file)\npython3 model.py --graph --output=graph.png (Image file)\n```\n\n## Documentation\n\nTo learn more about supported input and output types, dependency definition, task schedulers, result caching, etc. head over to the [package documentation](https://gitlab.com/geotom/modeltasks/-/wikis/Modeltasks)\n\n## Contribution\n\nPlease leave feedback, questions, suggestions on the [project's issue tracker](https://gitlab.com/geotom/modeltasks/-/issues).\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A lightweight workflow management system and task graph",
"version": "0.1.16",
"project_urls": {
"Homepage": "https://gitlab.com/geotom/modeltasks"
},
"split_keywords": [
"tasks",
" model",
" dag",
" graph",
" model",
" processing",
" workflow management"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "ec7171328df32c1d1b3671e8068d9185594cfcd71bc932e662b5142c2cb06db1",
"md5": "ec0b8b47eae95c7febbab5b539ec70ba",
"sha256": "7b6101c76b3aca933a6521668d5d4c5590b7208e90832cb188d036b88aed884d"
},
"downloads": -1,
"filename": "modeltasks-0.1.16-py3-none-any.whl",
"has_sig": false,
"md5_digest": "ec0b8b47eae95c7febbab5b539ec70ba",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": null,
"size": 39293,
"upload_time": "2024-03-28T12:48:41",
"upload_time_iso_8601": "2024-03-28T12:48:41.438998Z",
"url": "https://files.pythonhosted.org/packages/ec/71/71328df32c1d1b3671e8068d9185594cfcd71bc932e662b5142c2cb06db1/modeltasks-0.1.16-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-03-28 12:48:41",
"github": false,
"gitlab": true,
"bitbucket": false,
"codeberg": false,
"gitlab_user": "geotom",
"gitlab_project": "modeltasks",
"lcname": "modeltasks"
}
JSON