pypushflow

Name	pypushflow JSON
Version	0.7.0 JSON
	download
home_page	https://gitlab.esrf.fr/workflow/pypushflow/
Summary	A task scheduler for cyclic and acyclic graphs
upload_time	2024-11-13 21:33:12
maintainer	None
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": null,
    "docs_url": null,
    "requires_python": ">=3.6",
    "maintainer_email": null,
    "keywords": null,
    "author": "ESRF",
    "author_email": "svensson@esrf.fr",
    "download_url": "https://files.pythonhosted.org/packages/65/8b/008454418489877ee6ade57edd026b4ae6fa0342ecd643d9645410cac03c/pypushflow-0.7.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.7.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": "658b008454418489877ee6ade57edd026b4ae6fa0342ecd643d9645410cac03c",
                "md5": "34d711a5ad4c5ada9bf497a636ff2c83",
                "sha256": "8bbc45bcf3380ef60a9ae320a287d4828ded0d9f6fad6f66e7166f5393b97f69"
            },
            "downloads": -1,
            "filename": "pypushflow-0.7.0.tar.gz",
            "has_sig": false,
            "md5_digest": "34d711a5ad4c5ada9bf497a636ff2c83",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.6",
            "size": 32067,
            "upload_time": "2024-11-13T21:33:12",
            "upload_time_iso_8601": "2024-11-13T21:33:12.915429Z",
            "url": "https://files.pythonhosted.org/packages/65/8b/008454418489877ee6ade57edd026b4ae6fa0342ecd643d9645410cac03c/pypushflow-0.7.0.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2024-11-13 21:33:12",
    "github": false,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "lcname": "pypushflow"
}