# workflowlite
`pip install workflowlite`
A workflow engine that allows the execution of jobs composed of multiple steps. Each step is defined by an action
and a set of inputs and outputs.
The job definition can include `on_except` and `on_finish` hooks to handle the job's completion or failure.
The context is updated during execution with runtime information like `job_run_id` and `job_status`.
All the system variables are prefixed with `$`.
Example Usage:
```python
from workflowlite import WorkflowEngine, Context
engine = WorkflowEngine()
@engine.register_action('prepare')
def prepare(inputs: List[str], context: Context):
context['prepared_output'] = "prepared_output.mp4"
@engine.register_action('ffmpeg')
def ffmpeg(inputs: List[str], context: Context):
context['mp3_output'] = "output.mp3"
@engine.register_hook('on_finish')
def on_finish(context: Context):
print("Job finished successfully!")
@engine.register_hook('on_except')
def on_except(context: Context):
print("Job encountered an exception.")
job = {
"name": "video_processing",
"env": {"input_file": "video.mp4"},
"steps": [
{
"action": "prepare",
"input": ["input_file"],
"output": ["prepared_output"]
},
{
"action": "ffmpeg",
"input": ["prepared_output"],
"output": ["mp3_output"]
}
],
"output": ["mp3_output"],
"on_finish": "on_finish",
"on_except": "on_except"
}
with engine:
job_future = engine.submit_job(job)
try:
result = job_future.wait()
print(f"Job completed with output: {result}")
except Exception as e:
print(f"Job failed with exception: {e}")
```
Raw data
{
"_id": null,
"home_page": "https://github.com/c4pt0r/workflowlite",
"name": "workflowlite",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.6",
"maintainer_email": null,
"keywords": null,
"author": "Ed Huang",
"author_email": "dongxuhuang@yahoo.com",
"download_url": "https://files.pythonhosted.org/packages/38/cb/9ef42203abb69b87b76abf2ff257e3947c6f5416aaa6b5e3e381690546fe/workflowlite-0.2.3.tar.gz",
"platform": null,
"description": "# workflowlite\n\n`pip install workflowlite`\n\n A workflow engine that allows the execution of jobs composed of multiple steps. Each step is defined by an action\n and a set of inputs and outputs.\n\n The job definition can include `on_except` and `on_finish` hooks to handle the job's completion or failure.\n The context is updated during execution with runtime information like `job_run_id` and `job_status`.\n All the system variables are prefixed with `$`.\n\n Example Usage:\n\n ```python\n\nfrom workflowlite import WorkflowEngine, Context\n\nengine = WorkflowEngine()\n\n@engine.register_action('prepare')\ndef prepare(inputs: List[str], context: Context):\n context['prepared_output'] = \"prepared_output.mp4\"\n\n@engine.register_action('ffmpeg')\ndef ffmpeg(inputs: List[str], context: Context):\n context['mp3_output'] = \"output.mp3\"\n\n@engine.register_hook('on_finish')\ndef on_finish(context: Context):\n print(\"Job finished successfully!\")\n\n@engine.register_hook('on_except')\ndef on_except(context: Context):\n print(\"Job encountered an exception.\")\n\njob = {\n \"name\": \"video_processing\",\n \"env\": {\"input_file\": \"video.mp4\"},\n \"steps\": [\n {\n \"action\": \"prepare\",\n \"input\": [\"input_file\"],\n \"output\": [\"prepared_output\"]\n },\n {\n \"action\": \"ffmpeg\",\n \"input\": [\"prepared_output\"],\n \"output\": [\"mp3_output\"]\n }\n ],\n \"output\": [\"mp3_output\"],\n \"on_finish\": \"on_finish\",\n \"on_except\": \"on_except\"\n}\n\nwith engine:\n job_future = engine.submit_job(job)\n try:\n result = job_future.wait()\n print(f\"Job completed with output: {result}\")\n except Exception as e:\n print(f\"Job failed with exception: {e}\")\n```\n",
"bugtrack_url": null,
"license": null,
"summary": "a simple workflow engine",
"version": "0.2.3",
"project_urls": {
"Homepage": "https://github.com/c4pt0r/workflowlite"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "d9375c733ff35085cddc151d0a3eb359661d0973afe7897d2efb820cd14ba9f8",
"md5": "41b24a87bfa215652f4d619223f052ee",
"sha256": "e50dfc3104bba965375eda9f5ca4b3432436c0f4759d98b5ff3e9418a0234131"
},
"downloads": -1,
"filename": "workflowlite-0.2.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "41b24a87bfa215652f4d619223f052ee",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.6",
"size": 10223,
"upload_time": "2024-08-24T13:03:33",
"upload_time_iso_8601": "2024-08-24T13:03:33.079756Z",
"url": "https://files.pythonhosted.org/packages/d9/37/5c733ff35085cddc151d0a3eb359661d0973afe7897d2efb820cd14ba9f8/workflowlite-0.2.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "38cb9ef42203abb69b87b76abf2ff257e3947c6f5416aaa6b5e3e381690546fe",
"md5": "c5784815748b641749bf7d5215974349",
"sha256": "5007475d0b28024638d9cd7e86607130289083fd249cda47374cf7d041cb6fe8"
},
"downloads": -1,
"filename": "workflowlite-0.2.3.tar.gz",
"has_sig": false,
"md5_digest": "c5784815748b641749bf7d5215974349",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.6",
"size": 9492,
"upload_time": "2024-08-24T13:03:34",
"upload_time_iso_8601": "2024-08-24T13:03:34.936757Z",
"url": "https://files.pythonhosted.org/packages/38/cb/9ef42203abb69b87b76abf2ff257e3947c6f5416aaa6b5e3e381690546fe/workflowlite-0.2.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-08-24 13:03:34",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "c4pt0r",
"github_project": "workflowlite",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "workflowlite"
}
JSON
