{ "info": { "author": "Dimitry Ushakov", "author_email": "me@dimitryushakov.com", "bugtrack_url": null, "classifiers": [ "Development Status :: 3 - Alpha", "License :: OSI Approved :: MIT License", "Programming Language :: Python :: 3 :: Only", "Topic :: Software Development :: Libraries" ], "description": "# Workflow manager\n\nPython implementation of task-based workflow manager.\n\nThis package enables an easy wrap of any functionality that has dependencies on other functionality within your codebase.\n\nA simple use case would be a step by step wizard that has multiple success and failure scenarios. This package enables instantiation of all task rules in the wizard and then a simple manager wrapper to execute the workflow in one call.\n\nThe package also provides an ability to view the history of the workflow for debugging purposes.\n\n## How\n\nGiven the following business tasks:\n\n```\ntask one\ntask two\ntask three\ntask four\ntask five\ntask six\n```\n\nAnd the following business rules:\n\n```\nif task one succeeds, task two and task three will execute in sequence\nif task one fails, task five will execute\nif task two succeeds, task four will execute\nif task two fails, task five will execute\nif task three fails, task six will execute\nif task four fails, task six will execute\nif task five succeeds, task six will execute\n```\n\nThis module will set up a workflow that, based on status of the task, will execute the proper dependencies in the correct order.\n\nThe module will also short circuit any calls on failure scenarios but will execute all failure dependencies required to completely clean up your workflow.\n\nFor concrete examples, check out `tests/test_workflow.py`.\n\n## Setup\n\n`pip install workflow_manager`\n\nCreate your task, inhertit from `workflow_manager.task.Task` class, and overwrite the `execute` method with your own logic:\n\n```python\nfrom workflow_manager.task import Task\n\nclass CustomTask(Task):\n\n def __init__(self):\n super().__init__('my custom task')\n\n def execute(self, **kwargs):\n # your logic here\n if success:\n return (Task.success_state(), 'result', 'in', 'a', 'list')\n else:\n return (Task.failure_state(), 'this failed because of reasons')\n\n\nclass AnotherTask(Task):\n\n def __init__(self):\n super().__init__('some other task')\n\n def execute(self, **kwargs):\n # your logic here\n if success:\n return (Task.success_state(), 'result', 'in', 'a', 'list')\n else:\n return (Task.failure_state(), 'this failed because of reasons')\n```\n\nThen, add your business rules.\n\n```python\ncustomTask = CustomTask()\nanotherTask = AnotherTask()\ncustomTask.on_success(anotherTask, someOtherTask)\ncustomTask.on_failure(cleanupTask)\nanotherTask.on_success(keepItGoingTask)\nanotherTask.on_failure(cleanupTask)\n```\n\nYou can validate your workflow by printing your initial task (the one that will initiate the workflow):\n\n```python\n\nstr(customTask) # prints the entire workflow as json\ncustomTask.to_dict() # returns a dictionary of the workflow\n```\n\nFinally, simply register the initial task (the one that will initiate the workflow), and call `run` fuction:\n\n```python\nfrom workflow_manager.manager import Manager\n\n\nmanager = Manager()\nmanager.register_initial_task(customTask)\n\nmanager.run()\n```\n\nIf you want to see what happened after the workflow ends, you can call `show_executed_flow` method, which will return a list of tasks and the parameters.\n\n`manager.show_executed_flow()`\n", "description_content_type": null, "docs_url": null, "download_url": "", "downloads": { "last_day": -1, "last_month": -1, "last_week": -1 }, "home_page": "http://github.com/dimtruck/workflow-manager-py", "keywords": "", "license": "MIT", "maintainer": "", "maintainer_email": "", "name": "workflow_manager", "package_url": "https://pypi.org/project/workflow_manager/", "platform": "", "project_url": "https://pypi.org/project/workflow_manager/", "project_urls": { "Homepage": "http://github.com/dimtruck/workflow-manager-py" }, "release_url": "https://pypi.org/project/workflow_manager/0.4/", "requires_dist": null, "requires_python": "", "summary": "Wizard-link workflow manager", "version": "0.4" }, "last_serial": 3033007, "releases": { "0.2": [ { "comment_text": "", "digests": { "md5": "8e9854fa4cb1eef804071f077dda5ac6", "sha256": "f78a0983266a4cdf422e330e7663a5ab2a5405e735d8f2881356527a187b12ed" }, "downloads": -1, "filename": "workflow_manager-0.2.tar.gz", "has_sig": false, "md5_digest": "8e9854fa4cb1eef804071f077dda5ac6", "packagetype": "sdist", "python_version": "source", "requires_python": null, "size": 4234, "upload_time": "2017-07-18T15:24:34", "url": "https://files.pythonhosted.org/packages/8a/1a/b79a47c6aaac5a7cf0bfdb7a014b106e7494f8e0d331470e641e3d694bff/workflow_manager-0.2.tar.gz" } ], "0.3": [ { "comment_text": "", "digests": { "md5": "9da090cbce6dc0c95cbd0c2180caf443", "sha256": "17d3d2059c9fefb9d546ce95aeb16560b01e0b08f9bc9a6c05260aeec37193b0" }, "downloads": -1, "filename": "workflow_manager-0.3.tar.gz", "has_sig": false, "md5_digest": "9da090cbce6dc0c95cbd0c2180caf443", "packagetype": "sdist", "python_version": "source", "requires_python": null, "size": 4263, "upload_time": "2017-07-19T01:34:16", "url": "https://files.pythonhosted.org/packages/35/b2/cfc335613e4fd0093018cb223fb4d45e3a6ede09ad487c88a70fb094741d/workflow_manager-0.3.tar.gz" } ], "0.4": [ { "comment_text": "", "digests": { "md5": "c3398353626cb257acd83af391613545", "sha256": "481ee064e5abb4d2b5228377c5edc564246dcf92a9b07a5f4fcd5d8b52d2aa87" }, "downloads": -1, "filename": "workflow_manager-0.4.tar.gz", "has_sig": false, "md5_digest": "c3398353626cb257acd83af391613545", "packagetype": "sdist", "python_version": "source", "requires_python": null, "size": 4272, "upload_time": "2017-07-19T01:43:31", "url": "https://files.pythonhosted.org/packages/b6/98/eeef6938f55ce180b08db4e0fc0e4fa44573b8fd28657263ca3ee6964002/workflow_manager-0.4.tar.gz" } ] }, "urls": [ { "comment_text": "", "digests": { "md5": "c3398353626cb257acd83af391613545", "sha256": "481ee064e5abb4d2b5228377c5edc564246dcf92a9b07a5f4fcd5d8b52d2aa87" }, "downloads": -1, "filename": "workflow_manager-0.4.tar.gz", "has_sig": false, "md5_digest": "c3398353626cb257acd83af391613545", "packagetype": "sdist", "python_version": "source", "requires_python": null, "size": 4272, "upload_time": "2017-07-19T01:43:31", "url": "https://files.pythonhosted.org/packages/b6/98/eeef6938f55ce180b08db4e0fc0e4fa44573b8fd28657263ca3ee6964002/workflow_manager-0.4.tar.gz" } ] }