# A prefect extension giving prefect tasks super powers
## Motivation
Prefect tasks plus added functionality to enforce type checking and help in debugging helping to reduce negative engineering!
## How to setup
Using pip:
```bash
pip install supreme_task
```
## How to use
### Runtime type checking
Get runtime type checking thanks to [typeguard](http) by importing the `@task` decorator from `supreme_task` instead of `prefect`.
See the example `run.py` file:
```python run.py
from supreme_task import task
@task
def add(x: int, y: int) -> int:
return x + y
add.fn(x="1", y=2)
```
Running `python run.py` will raise the following exception:
```python
Traceback (most recent call last):
File "run.py", line 9, in <module>
add.fn(x="1", y=2)
File "run.py", line 5, in add
def add(x: int, y: int) -> int:
File "supreme-task-py38/lib/python3.8/site-packages/typeguard/_functions.py", line 135, in check_argument_types
check_type_internal(value, annotation, memo)
File "supreme-task-py38/lib/python3.8/site-packages/typeguard/_checkers.py", line 761, in check_type_internal
raise TypeCheckError(f"is not an instance of {qualified_name(origin_type)}")
typeguard.TypeCheckError: argument "x" (str) is not an instance of int
```
### Persistence of flow run inputs
Get persistence of flow run inputs by importing the `@task` decorator from `supreme_task` instead of `prefect` to help with debugging.
i.e. given a file `run.py`:
```python run.py
from supreme_task import task
from prefect import flow
from prefect.filesystems import LocalFileSystem
@task
def faulty_add(x: int, y: int) -> int:
if x == 1:
raise ValueError("x is 1")
return x + y
@flow(result_storage=LocalFileSystem(basepath="results/"))
def my_flow() -> None:
faulty_add(x=1, y=2)
my_flow()
```
We update our prefect configuration to enable result persistence:
```bash
prefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true
```
We then run the flow by running `python run.py`
We now inspect the results directory:
```bash
$ tree results
results
├── 514aaa4ae0134405a639cbd9a17365da
├── b662c63ff9854b0e9383d7f6cf0a5b76
└── inputs
└── faulty_add
└── 2023-06-10T10-47-04+0000
```
The inputs for failed task runs are saved under `results/inputs/<task_name>/<start_run_time>`.
Raw data
{
"_id": null,
"home_page": "https://github.com/marwan116/supreme-task/",
"name": "supreme-task",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8,<3.11",
"maintainer_email": "",
"keywords": "prefect,tasks,workflow,orchestration",
"author": "",
"author_email": "",
"download_url": "https://files.pythonhosted.org/packages/0b/74/ecd026c6e849affc2a9ad07f6a36aabfb477aa1938a75f1ca86e5d21c3dc/supreme_task-0.1.1.tar.gz",
"platform": null,
"description": "# A prefect extension giving prefect tasks super powers\n\n## Motivation\nPrefect tasks plus added functionality to enforce type checking and help in debugging helping to reduce negative engineering!\n\n## How to setup\n\nUsing pip:\n\n```bash\npip install supreme_task\n```\n\n## How to use\n\n### Runtime type checking\n\nGet runtime type checking thanks to [typeguard](http) by importing the `@task` decorator from `supreme_task` instead of `prefect`. \n\nSee the example `run.py` file:\n\n```python run.py\nfrom supreme_task import task\n\n@task\ndef add(x: int, y: int) -> int:\n return x + y\n\nadd.fn(x=\"1\", y=2)\n```\n\nRunning `python run.py` will raise the following exception:\n\n```python\nTraceback (most recent call last):\n File \"run.py\", line 9, in <module>\n add.fn(x=\"1\", y=2)\n File \"run.py\", line 5, in add\n def add(x: int, y: int) -> int:\n File \"supreme-task-py38/lib/python3.8/site-packages/typeguard/_functions.py\", line 135, in check_argument_types\n check_type_internal(value, annotation, memo)\n File \"supreme-task-py38/lib/python3.8/site-packages/typeguard/_checkers.py\", line 761, in check_type_internal\n raise TypeCheckError(f\"is not an instance of {qualified_name(origin_type)}\")\ntypeguard.TypeCheckError: argument \"x\" (str) is not an instance of int \n```\n\n### Persistence of flow run inputs\n\nGet persistence of flow run inputs by importing the `@task` decorator from `supreme_task` instead of `prefect` to help with debugging.\n\ni.e. given a file `run.py`:\n\n```python run.py\nfrom supreme_task import task\nfrom prefect import flow\nfrom prefect.filesystems import LocalFileSystem\n\n@task\ndef faulty_add(x: int, y: int) -> int:\n if x == 1:\n raise ValueError(\"x is 1\")\n return x + y\n\n@flow(result_storage=LocalFileSystem(basepath=\"results/\"))\ndef my_flow() -> None:\n faulty_add(x=1, y=2)\n\nmy_flow()\n```\n\nWe update our prefect configuration to enable result persistence:\n\n```bash\nprefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true\n```\n\nWe then run the flow by running `python run.py`\n\nWe now inspect the results directory:\n\n```bash\n$ tree results \nresults\n\u251c\u2500\u2500 514aaa4ae0134405a639cbd9a17365da\n\u251c\u2500\u2500 b662c63ff9854b0e9383d7f6cf0a5b76\n\u2514\u2500\u2500 inputs\n \u2514\u2500\u2500 faulty_add\n \u2514\u2500\u2500 2023-06-10T10-47-04+0000\n```\n\nThe inputs for failed task runs are saved under `results/inputs/<task_name>/<start_run_time>`.\n\n",
"bugtrack_url": null,
"license": "",
"summary": "Prefect tasks plus added functionality to enforce type checking and help in debugging helping to reduce negative engineering!",
"version": "0.1.1",
"project_urls": {
"Documentation": "https://github.com/marwan116/supreme-task/",
"Homepage": "https://github.com/marwan116/supreme-task/",
"Repository": "https://github.com/marwan116/supreme-task/"
},
"split_keywords": [
"prefect",
"tasks",
"workflow",
"orchestration"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "67f0506ed805bfe1bd2b7465ce9f24cdef28ebff240e42303267e7d2d8f86854",
"md5": "1441f69d60af3fed625436c728d280b9",
"sha256": "78549dc4fc96cbd2878bdaeca0a2bc1e780ca74a45d95ed3298f0b08dbe55639"
},
"downloads": -1,
"filename": "supreme_task-0.1.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "1441f69d60af3fed625436c728d280b9",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8,<3.11",
"size": 6827,
"upload_time": "2023-06-10T11:15:09",
"upload_time_iso_8601": "2023-06-10T11:15:09.272297Z",
"url": "https://files.pythonhosted.org/packages/67/f0/506ed805bfe1bd2b7465ce9f24cdef28ebff240e42303267e7d2d8f86854/supreme_task-0.1.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "0b74ecd026c6e849affc2a9ad07f6a36aabfb477aa1938a75f1ca86e5d21c3dc",
"md5": "186b3e9531bd7283642ec6ec77b0470f",
"sha256": "0da8e761a1db2012fb8450668ea925cef80803509cd05ed3deaac92a7ea666f2"
},
"downloads": -1,
"filename": "supreme_task-0.1.1.tar.gz",
"has_sig": false,
"md5_digest": "186b3e9531bd7283642ec6ec77b0470f",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8,<3.11",
"size": 5431,
"upload_time": "2023-06-10T11:15:11",
"upload_time_iso_8601": "2023-06-10T11:15:11.515461Z",
"url": "https://files.pythonhosted.org/packages/0b/74/ecd026c6e849affc2a9ad07f6a36aabfb477aa1938a75f1ca86e5d21c3dc/supreme_task-0.1.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-06-10 11:15:11",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "marwan116",
"github_project": "supreme-task",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "supreme-task"
}
JSON
