| Name | thecodecrate-pipeline JSON |
| Version |
1.5.0
JSON |
| download |
| home_page | None |
| Summary | This package provides a pipeline pattern implementation |
| upload_time | 2024-09-27 07:12:15 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.11 |
| license | The MIT License (MIT) Copyright (c) 2024 TheCodeCrate <loureiro.rg@gmail.com> Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
| keywords |
pipeline
python
library
python-pipeline
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Python-Pipeline
This package provides a pipeline pattern implementation.
It is a port of the [PHP League Pipeline](https://github.com/thephpleague/pipeline) package.
## Installation
```bash
pip install thecodecrate-pipeline
```
## Pipeline Pattern
The pipeline pattern allows you to easily compose sequential stages by
chaining stages.
In this particular implementation the interface consists of two parts:
- StageInterface
- PipelineInterface
A pipeline consists of zero, one, or multiple stages. A pipeline can process
a payload. During the processing the payload will be passed to the first stage.
From that moment on the resulting value is passed on from stage to stage.
In the simplest form, the execution chain can be represented as a foreach:
```python
result = payload
for stage in stages:
result = stage(result)
return result
```
Effectively this is the same as:
```python
result = stage3(stage2(stage1(payload)))
```
## Immutability
Pipelines are implemented as immutable stage chains. When you pipe a new
stage, a new pipeline will be created with the added stage. This makes
pipelines easy to reuse, and minimizes side-effects.
## Usage
Operations in a pipeline, stages, can be anything that satisfies the `Callable`
type-hint. So closures and anything that's invokable is good.
```python
pipeline = Pipeline().pipe(lambda payload: payload * 10)
# Returns 100
await pipeline.process(10)
```
## Type hinting
Use `Pipeline[PayloadType]` to type hint the payload type.
```python
pipeline = (
(Pipeline[int]())
.pipe(lambda payload: payload * 10)
)
# Returns 100
await pipeline.process(10)
```
## Class based stages
Class based stages are also possible. The `StageInterface[PayloadType]`
can be implemented which ensures you have the correct method signature
for the `__call__` method.
```python
class TimesTwoStage(StageInterface[int]):
async def __call__(self, payload: int) -> int:
return payload * 2
class AddOneStage(StageInterface[int]):
async def __call__(self, payload: int) -> int:
return payload + 1
pipeline = (
(Pipeline[int]())
.pipe(TimesTwoStage())
.pipe(AddOneStage())
)
# Returns 21
await pipeline.process(10)
```
## Re-usable Pipelines
Because the PipelineInterface is an extension of the StageInterface
pipelines can be re-used as stages. This creates a highly composable model
to create complex execution patterns while keeping the cognitive load low.
For example, if we'd want to compose a pipeline to process API calls, we'd create
something along these lines:
```python
process_api_request = (
(Pipeline())
.pipe(ExecuteHttpRequest())
.pipe(ParseJsonResponse())
)
pipeline = (
(Pipeline())
.pipe(ConvertToPsr7Request())
.pipe(process_api_request)
.pipe(ConvertToResponseDto())
)
await pipeline.process(DeleteBlogPost(post_id))
```
## Pipeline Builders
Because pipelines themselves are immutable, pipeline builders are introduced to
facilitate distributed composition of a pipeline.
The `PipelineBuilder[PayloadType]` builder collect stages and
allow you to create a pipeline at any given time.
```python
pipeline_builder = (
(PipelineBuilder())
.add(LogicalStage())
.add(AnotherStage())
.add(LastStage())
)
# Build the pipeline
pipeline = pipeline_builder.build()
```
## Declarative Pipeline Stages
You can define pipeline stages declaratively by specifying them as class-level attributes. This makes it easier to set up and reuse pipelines with predefined stages.
Example:
```python
class MyPipeline(Pipeline[int]):
processor_class = ChainedProcessor
stages = [
TimesTwoStage(),
TimesThreeStage(),
]
# Process the payload through the pipeline with the declared stages
result = await MyPipeline().process(5)
# Returns 30
```
In this example, `MyPipeline` declares its stages (`TimesTwoStage` and `TimesThreeStage`) directly in the class definition, making the pipeline easier to set up and more readable.
This declarative approach allows you to easily reuse pipelines across your project without needing to manually compose them every time.
## Exception handling
This package is completely transparent when dealing with exceptions. In no case
will this package catch an exception or silence an error. Exceptions should be
dealt with on a per-case basis. Either inside a __stage__ or at the time the
pipeline processes a payload.
```python
pipeline = Pipeline().pipe(lambda payload: payload / 0)
try:
await pipeline.process(10)
except ValueError as e:
# Handle the exception.
pass
```
Raw data
{
"_id": null,
"home_page": null,
"name": "thecodecrate-pipeline",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "pipeline, python, library, python-pipeline",
"author": null,
"author_email": "TheCodeCrate <loureiro.rg@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/58/3f/f46e5edb658243891200b71ff800a689fb25b1ac3a16cbc7899e47894283/thecodecrate_pipeline-1.5.0.tar.gz",
"platform": null,
"description": "# Python-Pipeline\n\nThis package provides a pipeline pattern implementation.\n\nIt is a port of the [PHP League Pipeline](https://github.com/thephpleague/pipeline) package.\n\n## Installation\n\n```bash\npip install thecodecrate-pipeline\n```\n\n## Pipeline Pattern\n\nThe pipeline pattern allows you to easily compose sequential stages by\nchaining stages.\n\nIn this particular implementation the interface consists of two parts:\n\n- StageInterface\n- PipelineInterface\n\nA pipeline consists of zero, one, or multiple stages. A pipeline can process\na payload. During the processing the payload will be passed to the first stage.\nFrom that moment on the resulting value is passed on from stage to stage.\n\nIn the simplest form, the execution chain can be represented as a foreach:\n\n```python\nresult = payload\n\nfor stage in stages:\n result = stage(result)\n\nreturn result\n```\n\nEffectively this is the same as:\n\n```python\nresult = stage3(stage2(stage1(payload)))\n```\n\n## Immutability\n\nPipelines are implemented as immutable stage chains. When you pipe a new\nstage, a new pipeline will be created with the added stage. This makes\npipelines easy to reuse, and minimizes side-effects.\n\n## Usage\n\nOperations in a pipeline, stages, can be anything that satisfies the `Callable`\ntype-hint. So closures and anything that's invokable is good.\n\n```python\npipeline = Pipeline().pipe(lambda payload: payload * 10)\n\n# Returns 100\nawait pipeline.process(10)\n```\n\n## Type hinting\n\nUse `Pipeline[PayloadType]` to type hint the payload type.\n\n```python\npipeline = (\n (Pipeline[int]())\n .pipe(lambda payload: payload * 10)\n)\n\n# Returns 100\nawait pipeline.process(10)\n```\n\n## Class based stages\n\nClass based stages are also possible. The `StageInterface[PayloadType]`\ncan be implemented which ensures you have the correct method signature\nfor the `__call__` method.\n\n```python\nclass TimesTwoStage(StageInterface[int]):\n async def __call__(self, payload: int) -> int:\n return payload * 2\n\nclass AddOneStage(StageInterface[int]):\n async def __call__(self, payload: int) -> int:\n return payload + 1\n\npipeline = (\n (Pipeline[int]())\n .pipe(TimesTwoStage())\n .pipe(AddOneStage())\n)\n\n# Returns 21\nawait pipeline.process(10)\n```\n\n## Re-usable Pipelines\n\nBecause the PipelineInterface is an extension of the StageInterface\npipelines can be re-used as stages. This creates a highly composable model\nto create complex execution patterns while keeping the cognitive load low.\n\nFor example, if we'd want to compose a pipeline to process API calls, we'd create\nsomething along these lines:\n\n```python\nprocess_api_request = (\n (Pipeline())\n .pipe(ExecuteHttpRequest())\n .pipe(ParseJsonResponse())\n)\n\npipeline = (\n (Pipeline())\n .pipe(ConvertToPsr7Request())\n .pipe(process_api_request)\n .pipe(ConvertToResponseDto())\n)\n\nawait pipeline.process(DeleteBlogPost(post_id))\n```\n\n## Pipeline Builders\n\nBecause pipelines themselves are immutable, pipeline builders are introduced to\nfacilitate distributed composition of a pipeline.\n\nThe `PipelineBuilder[PayloadType]` builder collect stages and\nallow you to create a pipeline at any given time.\n\n```python\npipeline_builder = (\n (PipelineBuilder())\n .add(LogicalStage())\n .add(AnotherStage())\n .add(LastStage())\n)\n\n# Build the pipeline\npipeline = pipeline_builder.build()\n```\n\n## Declarative Pipeline Stages\n\nYou can define pipeline stages declaratively by specifying them as class-level attributes. This makes it easier to set up and reuse pipelines with predefined stages.\n\nExample:\n\n```python\nclass MyPipeline(Pipeline[int]):\n processor_class = ChainedProcessor\n stages = [\n TimesTwoStage(),\n TimesThreeStage(),\n ]\n\n# Process the payload through the pipeline with the declared stages\nresult = await MyPipeline().process(5)\n\n# Returns 30\n```\n\nIn this example, `MyPipeline` declares its stages (`TimesTwoStage` and `TimesThreeStage`) directly in the class definition, making the pipeline easier to set up and more readable.\n\nThis declarative approach allows you to easily reuse pipelines across your project without needing to manually compose them every time.\n\n## Exception handling\n\nThis package is completely transparent when dealing with exceptions. In no case\nwill this package catch an exception or silence an error. Exceptions should be\ndealt with on a per-case basis. Either inside a __stage__ or at the time the\npipeline processes a payload.\n\n```python\npipeline = Pipeline().pipe(lambda payload: payload / 0)\n\ntry:\n await pipeline.process(10)\nexcept ValueError as e:\n # Handle the exception.\n pass\n```\n",
"bugtrack_url": null,
"license": "The MIT License (MIT) Copyright (c) 2024 TheCodeCrate <loureiro.rg@gmail.com> Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.",
"summary": "This package provides a pipeline pattern implementation",
"version": "1.5.0",
"project_urls": {
"documentation": "https://github.com/thecodecrate/python-pipeline",
"repository": "https://github.com/thecodecrate/python-pipeline"
},
"split_keywords": [
"pipeline",
" python",
" library",
" python-pipeline"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "eb12e9a66de5bd1eb8db9cdb4832cee4fdfd0efade6051b5872f2a026a803876",
"md5": "32668de199670acedae993def263fd8c",
"sha256": "acdfafd351db03b40f4c5b3ea34affada483117a01ee7e3a43a85c797330038c"
},
"downloads": -1,
"filename": "thecodecrate_pipeline-1.5.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "32668de199670acedae993def263fd8c",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 13666,
"upload_time": "2024-09-27T07:12:13",
"upload_time_iso_8601": "2024-09-27T07:12:13.112028Z",
"url": "https://files.pythonhosted.org/packages/eb/12/e9a66de5bd1eb8db9cdb4832cee4fdfd0efade6051b5872f2a026a803876/thecodecrate_pipeline-1.5.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "583ff46e5edb658243891200b71ff800a689fb25b1ac3a16cbc7899e47894283",
"md5": "13bae93aa6415f9d9627fabcbb249c37",
"sha256": "b04144c6eb881b9c5a95ee567b9da32085462502ac7c33a940bae4cf6ff72bc9"
},
"downloads": -1,
"filename": "thecodecrate_pipeline-1.5.0.tar.gz",
"has_sig": false,
"md5_digest": "13bae93aa6415f9d9627fabcbb249c37",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 10935,
"upload_time": "2024-09-27T07:12:15",
"upload_time_iso_8601": "2024-09-27T07:12:15.260981Z",
"url": "https://files.pythonhosted.org/packages/58/3f/f46e5edb658243891200b71ff800a689fb25b1ac3a16cbc7899e47894283/thecodecrate_pipeline-1.5.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-09-27 07:12:15",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "thecodecrate",
"github_project": "python-pipeline",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "thecodecrate-pipeline"
}
