{ "info": { "author": "Nino Walker", "author_email": "nino.walker@gmail.com", "bugtrack_url": null, "classifiers": [ "License :: OSI Approved :: BSD License" ], "description": "marx\n=====\n\nFramework/tools for python that get work done, make code easy to test, \nand turn spaghetti logic back into something you can love.\n\nThe modeling here provides clear contracts, state encapsulation,\nand strict typing.\n\n\nExample\n-------\n\nYou'll find the [full example](./tests/workflow/example_1.py) in the tests/ directory.\n\n \"\"\"\n This defines a workflow around the imperative: \"Throw a Pie\".\n \n In a simple system this is about 5 lines of code. But in others,\n where business logic requires the interaction of many systems,\n logic is smeared around and duplicated, and not well encapsulated.\n \n The approach here splits execution into discrete units which can be\n composed into other workflows encouraging reuse, testability, etc.\n \"\"\"\n \n \n from marx.workflow.step import LogicUnit, ArgSpec, Step, ResultSpec\n from marx.workflow.flow import Workflow\n from marx.workflow.context import DefaultContext, Field\n from tests.workflow.example_objects import User, PermissionDeniedError\n \n \n class IsUserAuthorized(LogicUnit):\n \"\"\"Checks permission for a user+action, and notifies authorities\n if it fails.\"\"\"\n \n user = ArgSpec(User, docs=\"The user performing the action\")\n \n def __init__(self, action):\n \"\"\"\n @param action: The action that will be checked.\n \"\"\"\n self.action = action\n \n def __call__(self, user):\n \"\"\"\n @param user: The user performing the action.\n \"\"\"\n if self.is_authorized(user):\n return\n self.notify_authorities(user, self.action)\n raise PermissionDeniedError(self.action)\n \n def is_authorized(self, user):\n return user.name in (\"bob\", \"mary\")\n \n def notify_authorities(self, user, action):\n print \"AUTHORITIES!!!\", user, \" attempted illegal action\", self.action\n \n \n class MakePie(LogicUnit):\n \"\"\"Makes the pie.\"\"\"\n \n maker = ArgSpec(User, docs=\"The person making pie.\")\n pie = ResultSpec(basestring, docs=\"Kind of pie\")\n \n def __call__(self, maker):\n maker.increment(\"pies_made\", 1)\n self.result.pie = 'lemon'\n \n \n class ThrowThing(LogicUnit):\n \"\"\"Subject Object (Verb) Indirect-object\"\"\"\n \n actor = ArgSpec(User)\n hit = ResultSpec(bool, default=False, docs=\"Did we get 'em?\")\n \n # we omit target and thing here, because we don't\n # need to enumerate/type constrain the values in this example\n \n def __call__(self, actor, thing, target):\n actor.increment(\"things_throw\")\n print \"Throwing\", thing\n self.result.hit = actor.can_throw()\n # we don't need to return, but we can.\n return self.result\n \n \n class ThrowPieContext(DefaultContext):\n \"\"\"The execution context for the ThrowPieWorkflow.\"\"\"\n thrower = Field(User, docs=\"Somebody has to throw it\")\n target = Field(User, docs=\"At somebody\")\n pie = Field(str, docs=\"A pie, which we make along the way\")\n was_hit = Field(bool, docs=\"Success of the throwing event\")\n \n \"\"\" A workflow is a series of steps.\"\"\"\n IsUserAuthorizedStep = Step(\n IsUserAuthorized(\"throw_pie\"),\n # we bind from the context to the arguments of the method.\n arg_map={IsUserAuthorized.USER: ThrowPieContext.THROWER}\n )\n MakePieStep = Step(\n MakePie(),\n arg_map={MakePie.MAKER: ThrowPieContext.THROWER},\n # we bind from the returned result back to the context\n result_map=MakePie.ResultMap(ThrowPieContext)\n )\n ThrowThingStep = Step(\n ThrowThing(),\n arg_map=ThrowThing.AutoMap({ThrowThing.ACTOR: ThrowPieContext.THROWER,\n ThrowThing.THING: ThrowPieContext.PIE}),\n result_map={ThrowPieContext.WAS_HIT: 'hit'}\n )\n \n \n \"\"\" There are a few ways to build up a workflow. By constructor...\"\"\"\n ThrowPieWorkflowA = Workflow(steps=[\n IsUserAuthorizedStep,\n MakePieStep,\n ThrowThingStep,\n ])\n \n \"\"\" Or using add_step...\"\"\"\n ThrowPieWorkflowB = Workflow().add_step(\n IsUserAuthorizedStep\n ).add_step(\n MakePieStep\n ).add_step(\n ThrowThingStep\n )\n \n \"\"\" Or using the overloaded addition operator.\"\"\"\n EmptyWorkflow = Workflow()\n ThrowPieWorkflowC_A = EmptyWorkflow + IsUserAuthorizedStep + MakePieStep\n ThrowPieWorkflowC_B = EmptyWorkflow + ThrowThingStep\n ThrowPieWorkflowC = ThrowPieWorkflowC_A + ThrowPieWorkflowC_B\n \n \n def run():\n \"\"\"To execute a workflow, prepare a context, and pass it through.\"\"\"\n ctx = ThrowPieContext()\n ctx.thrower = User(\"bob\")\n ctx.target = User(\"frank\")\n for WorkflowType in (ThrowPieWorkflowA,\n ThrowPieWorkflowB,\n ThrowPieWorkflowC):\n try:\n WorkflowType(ctx)\n assert ctx.was_hit is not None\n assert ctx.pie == 'lemon'\n return ctx\n except PermissionDeniedError:\n assert False\n \n # Ensure that our ThrowPieWorkflowC did not modify its components\n assert EmptyWorkflow.steps == []\n assert ThrowThingStep not in ThrowPieWorkflowC_A.steps", "description_content_type": null, "docs_url": null, "download_url": "UNKNOWN", "downloads": { "last_day": -1, "last_month": -1, "last_week": -1 }, "home_page": "https://github.com/ninowalker/marx", "keywords": null, "license": "BSD", "maintainer": null, "maintainer_email": null, "name": "marx-workflows", "package_url": "https://pypi.org/project/marx-workflows/", "platform": "UNKNOWN", "project_url": "https://pypi.org/project/marx-workflows/", "project_urls": { "Download": "UNKNOWN", "Homepage": "https://github.com/ninowalker/marx" }, "release_url": "https://pypi.org/project/marx-workflows/0.0.8/", "requires_dist": null, "requires_python": null, "summary": "marx", "version": "0.0.8" }, "last_serial": 1282554, "releases": { "0.0.8": [ { "comment_text": "", "digests": { "md5": "468244653062674dce09dce5e22812fb", "sha256": "69cbd506276c613d7f1d766eff3182597d7d8c1d989571a435e3cb5c77833039" }, "downloads": -1, "filename": "marx-workflows-0.0.8.tar.gz", "has_sig": false, "md5_digest": "468244653062674dce09dce5e22812fb", "packagetype": "sdist", "python_version": "source", "requires_python": null, "size": 19607, "upload_time": "2014-10-25T16:13:22", "url": "https://files.pythonhosted.org/packages/e1/e3/9a5a9a1c67edd2359f5a2782a533b92d256faa0835aee34baff242f6625b/marx-workflows-0.0.8.tar.gz" } ] }, "urls": [ { "comment_text": "", "digests": { "md5": "468244653062674dce09dce5e22812fb", "sha256": "69cbd506276c613d7f1d766eff3182597d7d8c1d989571a435e3cb5c77833039" }, "downloads": -1, "filename": "marx-workflows-0.0.8.tar.gz", "has_sig": false, "md5_digest": "468244653062674dce09dce5e22812fb", "packagetype": "sdist", "python_version": "source", "requires_python": null, "size": 19607, "upload_time": "2014-10-25T16:13:22", "url": "https://files.pythonhosted.org/packages/e1/e3/9a5a9a1c67edd2359f5a2782a533b92d256faa0835aee34baff242f6625b/marx-workflows-0.0.8.tar.gz" } ] }