pypushflow

Name	pypushflow JSON
Version	0.6.0 JSON
	download
home_page	https://gitlab.esrf.fr/workflow/pypushflow/
Summary	A task scheduler for cyclic and acyclic graphs
upload_time	2023-09-07 11:50:43
maintainer
docs_url	None
author	ESRF
requires_python	>=3.6
license	MIT
keywords
VCS
bugtrack_url
requirements	No requirements were recorded.
Travis-CI	No Travis.
coveralls test coverage	No coveralls.

            # pypushflow

A task scheduler for cyclic and acyclic graphs

## Install

```bash
pip install pypushflow[mx]
```

Use the `mx` option for installation at MX beamlines.

## Run tests

```bash
pip install pypushflow[test]
pytest
```

## Getting started

```python
import logging
from pypushflow.Workflow import Workflow
from pypushflow.StopActor import StopActor
from pypushflow.StartActor import StartActor
from pypushflow.PythonActor import PythonActor
from pypushflow.ThreadCounter import ThreadCounter


class MyWorkflow(Workflow):
    def __init__(self, name):
        super().__init__(name, level=logging.DEBUG)
        ctr = ThreadCounter(parent=self)
        self.startActor = StartActor(parent=self, thread_counter=ctr)
        self.pythonActor = PythonActor(
            parent=self,
            script="pypushflow.tests.tasks.pythonActorTest.py",
            name="Python Actor Test",
            thread_counter=ctr,
        )
        self.stopActor = StopActor(parent=self, thread_counter=ctr)
        self.startActor.connect(self.pythonActor)
        self.pythonActor.connect(self.stopActor)


testMyWorkflow = MyWorkflow("Test workflow")
inData = {"name": "World"}
outData = testMyWorkflow.run(inData, timeout=15, pool_type="process")
assert outData["reply"] == "Hello World!"
```

## Documentation

https://pypushflow.readthedocs.io/

            

Raw data

            {
    "_id": null,
    "home_page": "https://gitlab.esrf.fr/workflow/pypushflow/",
    "name": "pypushflow",
    "maintainer": "",
    "docs_url": null,
    "requires_python": ">=3.6",
    "maintainer_email": "",
    "keywords": "",
    "author": "ESRF",
    "author_email": "svensson@esrf.fr",
    "download_url": "https://files.pythonhosted.org/packages/ea/f4/46d6d36543a255c5c5fbd521efb783b513f52ecc6e4fe544710cd9b1357d/pypushflow-0.6.0.tar.gz",
    "platform": null,
    "description": "# pypushflow\n\nA task scheduler for cyclic and acyclic graphs\n\n## Install\n\n```bash\npip install pypushflow[mx]\n```\n\nUse the `mx` option for installation at MX beamlines.\n\n## Run tests\n\n```bash\npip install pypushflow[test]\npytest\n```\n\n## Getting started\n\n```python\nimport logging\nfrom pypushflow.Workflow import Workflow\nfrom pypushflow.StopActor import StopActor\nfrom pypushflow.StartActor import StartActor\nfrom pypushflow.PythonActor import PythonActor\nfrom pypushflow.ThreadCounter import ThreadCounter\n\n\nclass MyWorkflow(Workflow):\n    def __init__(self, name):\n        super().__init__(name, level=logging.DEBUG)\n        ctr = ThreadCounter(parent=self)\n        self.startActor = StartActor(parent=self, thread_counter=ctr)\n        self.pythonActor = PythonActor(\n            parent=self,\n            script=\"pypushflow.tests.tasks.pythonActorTest.py\",\n            name=\"Python Actor Test\",\n            thread_counter=ctr,\n        )\n        self.stopActor = StopActor(parent=self, thread_counter=ctr)\n        self.startActor.connect(self.pythonActor)\n        self.pythonActor.connect(self.stopActor)\n\n\ntestMyWorkflow = MyWorkflow(\"Test workflow\")\ninData = {\"name\": \"World\"}\noutData = testMyWorkflow.run(inData, timeout=15, pool_type=\"process\")\nassert outData[\"reply\"] == \"Hello World!\"\n```\n\n## Documentation\n\nhttps://pypushflow.readthedocs.io/\n",
    "bugtrack_url": null,
    "license": "MIT",
    "summary": "A task scheduler for cyclic and acyclic graphs",
    "version": "0.6.0",
    "project_urls": {
        "Documentation": "https://pypushflow.readthedocs.io/",
        "Homepage": "https://gitlab.esrf.fr/workflow/pypushflow/",
        "Source": "https://gitlab.esrf.fr/workflow/pypushflow/",
        "Tracker": "https://gitlab.esrf.fr/workflow/pypushflow/issues/"
    },
    "split_keywords": [],
    "urls": [
        {
            "comment_text": "",
            "digests": {
                "blake2b_256": "eaf446d6d36543a255c5c5fbd521efb783b513f52ecc6e4fe544710cd9b1357d",
                "md5": "7211dac60a510e2e2b33517797ca4cec",
                "sha256": "775c6384e29989bfb00658f00c9970e43ed225148353ac8e27b0ada4240e8138"
            },
            "downloads": -1,
            "filename": "pypushflow-0.6.0.tar.gz",
            "has_sig": false,
            "md5_digest": "7211dac60a510e2e2b33517797ca4cec",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.6",
            "size": 28090,
            "upload_time": "2023-09-07T11:50:43",
            "upload_time_iso_8601": "2023-09-07T11:50:43.568180Z",
            "url": "https://files.pythonhosted.org/packages/ea/f4/46d6d36543a255c5c5fbd521efb783b513f52ecc6e4fe544710cd9b1357d/pypushflow-0.6.0.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2023-09-07 11:50:43",
    "github": false,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "lcname": "pypushflow"
}