| Name | laygo JSON |
| Version |
0.1.2
JSON |
| download |
| home_page | None |
| Summary | A lightweight Python library for building resilient, in-memory data pipelines with elegant, chainable syntax |
| upload_time | 2025-07-16 19:41:27 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.12 |
| license | MIT |
| keywords |
chaining
data-processing
etl
functional
pipeline
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
<!-- PROJECT_TITLE -->
# Laygo
<!-- PROJECT_TAGLINE -->
**A lightweight Python library for building resilient, in-memory data pipelines with elegant, chainable syntax.**
<!-- BADGES_SECTION -->
[](https://opensource.org/licenses/MIT)
[](https://www.python.org/downloads/)
[](https://github.com/astral-sh/uv)
[](https://github.com/astral-sh/ruff)
---
## 🎯 Overview
**Laygo** is a lightweight Python library for building resilient, in-memory data pipelines. It provides a fluent API to layer transformations, manage context, and handle errors with elegant, chainable syntax.
**Key Features:**
- **Fluent API**: Chainable method syntax for readable data transformations
- **Performance Optimized**: Uses chunked processing and list comprehensions for maximum speed
- **Memory Efficient**: Lazy evaluation and streaming support for large datasets
- **Parallel Processing**: Built-in ThreadPoolExecutor for CPU-intensive operations
- **Context Management**: Shared state across pipeline operations for stateful processing
- **Error Handling**: Comprehensive error handling
- **Type Safety**: Full type hints support with generic types
---
## 📦 Installation
```bash
pip install laygo
```
Or for development:
```bash
git clone https://github.com/ringoldsdev/laygo-python.git
cd laygo-python
pip install -e ".[dev]"
```
### 🐳 Dev Container Setup
If you're using this project in a dev container, you'll need to configure Git to use HTTPS instead of SSH for authentication:
```bash
# Switch to HTTPS remote URL
git remote set-url origin https://github.com/ringoldsdev/laygo-python.git
# Configure Git to use HTTPS for all GitHub operations
git config --global url."https://github.com/".insteadOf "git@github.com:"
```
---
## ▶️ Usage
### Basic Pipeline Operations
```python
from laygo import Pipeline
# Simple data transformation
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
result = (
Pipeline(data)
.transform(lambda t: t.filter(lambda x: x % 2 == 0)) # Keep even numbers
.transform(lambda t: t.map(lambda x: x * 2)) # Double them
.to_list()
)
print(result) # [4, 8, 12, 16, 20]
```
### Context-Aware Operations
```python
from laygo import Pipeline
from laygo import PipelineContext
# Create context with shared state
context: PipelineContext = {"multiplier": 3, "threshold": 10}
result = (
Pipeline([1, 2, 3, 4, 5])
.context(context)
.transform(lambda t: t.map(lambda x, ctx: x * ctx["multiplier"]))
.transform(lambda t: t.filter(lambda x, ctx: x > ctx["threshold"]))
.to_list()
)
print(result) # [12, 15]
```
### ETL Pipeline Example
```python
from laygo import Pipeline
# Sample employee data processing
employees = [
{"name": "Alice", "age": 25, "salary": 50000},
{"name": "Bob", "age": 30, "salary": 60000},
{"name": "Charlie", "age": 35, "salary": 70000},
{"name": "David", "age": 28, "salary": 55000},
]
# Extract, Transform, Load pattern
high_earners = (
Pipeline(employees)
.transform(lambda t: t.filter(lambda emp: emp["age"] > 28)) # Extract
.transform(lambda t: t.map(lambda emp: { # Transform
"name": emp["name"],
"annual_salary": emp["salary"],
"monthly_salary": emp["salary"] / 12
}))
.transform(lambda t: t.filter(lambda emp: emp["annual_salary"] > 55000)) # Filter
.to_list()
)
```
### Using Transformers Directly
```python
from laygo import Transformer
# Create a reusable transformation pipeline
transformer = (
Transformer.init(int)
.filter(lambda x: x % 2 == 0) # Keep even numbers
.map(lambda x: x * 2) # Double them
.filter(lambda x: x > 5) # Keep > 5
)
# Apply to different datasets
result1 = list(transformer([1, 2, 3, 4, 5])) # [4, 8]
result2 = list(transformer(range(10))) # [4, 8, 12, 16, 20]
```
### Custom Transformer Composition
```python
from laygo import Pipeline
from laygo import Transformer
# Create reusable transformation components
validate_data = Transformer.init(dict).filter(lambda x: x.get("id") is not None)
normalize_text = Transformer.init(dict).map(lambda x: {**x, "name": x["name"].strip().title()})
# Use transformers directly with Pipeline.transform()
result = (
Pipeline(raw_data)
.transform(validate_data) # Pass transformer directly
.transform(normalize_text) # Pass transformer directly
.to_list()
)
```
### Parallel Processing
```python
from laygo import Pipeline
from laygo import ParallelTransformer
# Process large datasets with multiple threads
large_data = range(100_000)
# Create parallel transformer
parallel_processor = (
ParallelTransformer.init(
int,
max_workers=4,
ordered=True, # Maintain result order
chunk_size=10000 # Process in chunks
).map(lambda x: x ** 2)
)
results = (
Pipeline(large_data)
.transform(parallel_processor)
.transform(lambda t: t.filter(lambda x: x > 100))
.first(1000) # Get first 1000 results
)
```
### Error Handling and Recovery
```python
from laygo import Pipeline
from laygo import Transformer
def risky_operation(x):
if x == 5:
raise ValueError("Cannot process 5")
return x * 2
def error_handler(chunk, error, context):
print(f"Error in chunk {chunk}: {error}")
return [0] * len(chunk) # Return default values
# Pipeline with error recovery
result = (
Pipeline([1, 2, 3, 4, 5, 6])
.transform(lambda t: t.map(risky_operation).catch(
lambda sub_t: sub_t.map(lambda x: x + 1),
on_error=error_handler
))
.to_list()
)
```
---
## ⚙️ Projects using Laygo
- **[Efemel](https://github.com/ringoldsdev/efemel)** - A CLI tool that processes Python files as configuration markup and exports them to JSON/YAML, replacing traditional templating DSLs with native Python syntax.
---
## 📄 License
This project is licensed under the **MIT License** - see the [LICENSE](LICENSE) file for details.
---
## 🚀 Built With
- **[Python 3.12+](https://python.org)** - Core language with modern type hints
- **[Ruff](https://github.com/astral-sh/ruff)** - Code formatting and linting
- **[Pytest](https://pytest.org/)** - Testing framework
- **[DevContainers](https://containers.dev/)** - Consistent development environment
- **[GitHub Actions](https://github.com/features/actions)** - CI/CD automation
---
**⭐ Star this repository if Laygo helps your data processing workflows!**
Raw data
{
"_id": null,
"home_page": null,
"name": "laygo",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": null,
"keywords": "chaining, data-processing, etl, functional, pipeline",
"author": null,
"author_email": "Ringolds Lescinskis <ringolds@lescinskis.com>",
"download_url": "https://files.pythonhosted.org/packages/d7/1a/1832504b8a826f4b5543490a4ef97f69df97f9b165f76402213c692a55be/laygo-0.1.2.tar.gz",
"platform": null,
"description": "<!-- PROJECT_TITLE -->\n\n# Laygo\n\n<!-- PROJECT_TAGLINE -->\n\n**A lightweight Python library for building resilient, in-memory data pipelines with elegant, chainable syntax.**\n\n<!-- BADGES_SECTION -->\n\n[](https://opensource.org/licenses/MIT)\n[](https://www.python.org/downloads/)\n[](https://github.com/astral-sh/uv)\n[](https://github.com/astral-sh/ruff)\n\n---\n\n## \ud83c\udfaf Overview\n\n**Laygo** is a lightweight Python library for building resilient, in-memory data pipelines. It provides a fluent API to layer transformations, manage context, and handle errors with elegant, chainable syntax.\n\n**Key Features:**\n\n- **Fluent API**: Chainable method syntax for readable data transformations\n- **Performance Optimized**: Uses chunked processing and list comprehensions for maximum speed\n- **Memory Efficient**: Lazy evaluation and streaming support for large datasets\n- **Parallel Processing**: Built-in ThreadPoolExecutor for CPU-intensive operations\n- **Context Management**: Shared state across pipeline operations for stateful processing\n- **Error Handling**: Comprehensive error handling\n- **Type Safety**: Full type hints support with generic types\n\n---\n\n## \ud83d\udce6 Installation\n\n```bash\npip install laygo\n```\n\nOr for development:\n\n```bash\ngit clone https://github.com/ringoldsdev/laygo-python.git\ncd laygo-python\npip install -e \".[dev]\"\n```\n\n### \ud83d\udc33 Dev Container Setup\n\nIf you're using this project in a dev container, you'll need to configure Git to use HTTPS instead of SSH for authentication:\n\n```bash\n# Switch to HTTPS remote URL\ngit remote set-url origin https://github.com/ringoldsdev/laygo-python.git\n\n# Configure Git to use HTTPS for all GitHub operations\ngit config --global url.\"https://github.com/\".insteadOf \"git@github.com:\"\n```\n\n---\n\n## \u25b6\ufe0f Usage\n\n### Basic Pipeline Operations\n\n```python\nfrom laygo import Pipeline\n\n# Simple data transformation\ndata = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]\nresult = (\n Pipeline(data)\n .transform(lambda t: t.filter(lambda x: x % 2 == 0)) # Keep even numbers\n .transform(lambda t: t.map(lambda x: x * 2)) # Double them\n .to_list()\n)\nprint(result) # [4, 8, 12, 16, 20]\n```\n\n### Context-Aware Operations\n\n```python\nfrom laygo import Pipeline\nfrom laygo import PipelineContext\n\n# Create context with shared state\ncontext: PipelineContext = {\"multiplier\": 3, \"threshold\": 10}\n\nresult = (\n Pipeline([1, 2, 3, 4, 5])\n .context(context)\n .transform(lambda t: t.map(lambda x, ctx: x * ctx[\"multiplier\"]))\n .transform(lambda t: t.filter(lambda x, ctx: x > ctx[\"threshold\"]))\n .to_list()\n)\nprint(result) # [12, 15]\n```\n\n### ETL Pipeline Example\n\n```python\nfrom laygo import Pipeline\n\n# Sample employee data processing\nemployees = [\n {\"name\": \"Alice\", \"age\": 25, \"salary\": 50000},\n {\"name\": \"Bob\", \"age\": 30, \"salary\": 60000},\n {\"name\": \"Charlie\", \"age\": 35, \"salary\": 70000},\n {\"name\": \"David\", \"age\": 28, \"salary\": 55000},\n]\n\n# Extract, Transform, Load pattern\nhigh_earners = (\n Pipeline(employees)\n .transform(lambda t: t.filter(lambda emp: emp[\"age\"] > 28)) # Extract\n .transform(lambda t: t.map(lambda emp: { # Transform\n \"name\": emp[\"name\"],\n \"annual_salary\": emp[\"salary\"],\n \"monthly_salary\": emp[\"salary\"] / 12\n }))\n .transform(lambda t: t.filter(lambda emp: emp[\"annual_salary\"] > 55000)) # Filter\n .to_list()\n)\n```\n\n### Using Transformers Directly\n\n```python\nfrom laygo import Transformer\n\n# Create a reusable transformation pipeline\ntransformer = (\n Transformer.init(int)\n .filter(lambda x: x % 2 == 0) # Keep even numbers\n .map(lambda x: x * 2) # Double them\n .filter(lambda x: x > 5) # Keep > 5\n)\n\n# Apply to different datasets\nresult1 = list(transformer([1, 2, 3, 4, 5])) # [4, 8]\nresult2 = list(transformer(range(10))) # [4, 8, 12, 16, 20]\n```\n\n### Custom Transformer Composition\n\n```python\nfrom laygo import Pipeline\nfrom laygo import Transformer\n\n# Create reusable transformation components\nvalidate_data = Transformer.init(dict).filter(lambda x: x.get(\"id\") is not None)\nnormalize_text = Transformer.init(dict).map(lambda x: {**x, \"name\": x[\"name\"].strip().title()})\n\n# Use transformers directly with Pipeline.transform()\nresult = (\n Pipeline(raw_data)\n .transform(validate_data) # Pass transformer directly\n .transform(normalize_text) # Pass transformer directly\n .to_list()\n)\n```\n\n### Parallel Processing\n\n```python\nfrom laygo import Pipeline\nfrom laygo import ParallelTransformer\n\n# Process large datasets with multiple threads\nlarge_data = range(100_000)\n\n# Create parallel transformer\nparallel_processor = (\n ParallelTransformer.init(\n int,\n max_workers=4,\n ordered=True, # Maintain result order\n chunk_size=10000 # Process in chunks\n ).map(lambda x: x ** 2)\n)\n\nresults = (\n Pipeline(large_data)\n .transform(parallel_processor)\n .transform(lambda t: t.filter(lambda x: x > 100))\n .first(1000) # Get first 1000 results\n)\n```\n\n### Error Handling and Recovery\n\n```python\nfrom laygo import Pipeline\nfrom laygo import Transformer\n\ndef risky_operation(x):\n if x == 5:\n raise ValueError(\"Cannot process 5\")\n return x * 2\n\ndef error_handler(chunk, error, context):\n print(f\"Error in chunk {chunk}: {error}\")\n return [0] * len(chunk) # Return default values\n\n# Pipeline with error recovery\nresult = (\n Pipeline([1, 2, 3, 4, 5, 6])\n .transform(lambda t: t.map(risky_operation).catch(\n lambda sub_t: sub_t.map(lambda x: x + 1),\n on_error=error_handler\n ))\n .to_list()\n)\n```\n\n---\n\n## \u2699\ufe0f Projects using Laygo\n\n- **[Efemel](https://github.com/ringoldsdev/efemel)** - A CLI tool that processes Python files as configuration markup and exports them to JSON/YAML, replacing traditional templating DSLs with native Python syntax.\n\n---\n\n## \ud83d\udcc4 License\n\nThis project is licensed under the **MIT License** - see the [LICENSE](LICENSE) file for details.\n\n---\n\n## \ud83d\ude80 Built With\n\n- **[Python 3.12+](https://python.org)** - Core language with modern type hints\n- **[Ruff](https://github.com/astral-sh/ruff)** - Code formatting and linting\n- **[Pytest](https://pytest.org/)** - Testing framework\n- **[DevContainers](https://containers.dev/)** - Consistent development environment\n- **[GitHub Actions](https://github.com/features/actions)** - CI/CD automation\n\n---\n\n**\u2b50 Star this repository if Laygo helps your data processing workflows!**\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A lightweight Python library for building resilient, in-memory data pipelines with elegant, chainable syntax",
"version": "0.1.2",
"project_urls": {
"Documentation": "https://github.com/ringoldsdev/laygo-python/wiki",
"Homepage": "https://github.com/ringoldsdev/laygo-python",
"Issues": "https://github.com/ringoldsdev/laygo-python/issues",
"Repository": "https://github.com/ringoldsdev/laygo-python.git"
},
"split_keywords": [
"chaining",
" data-processing",
" etl",
" functional",
" pipeline"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "7e1f009dad16fdcaba372cd36c944a753eac6dc3b32d8e58b4c9c3392123e36f",
"md5": "0384e6e4a95bbf40055e17ed99df068e",
"sha256": "a824e2b469c9d374a28c7e2e8da984a6cad9336871d63b758c477520fa6db177"
},
"downloads": -1,
"filename": "laygo-0.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0384e6e4a95bbf40055e17ed99df068e",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 12063,
"upload_time": "2025-07-16T19:41:26",
"upload_time_iso_8601": "2025-07-16T19:41:26.327520Z",
"url": "https://files.pythonhosted.org/packages/7e/1f/009dad16fdcaba372cd36c944a753eac6dc3b32d8e58b4c9c3392123e36f/laygo-0.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "d71a1832504b8a826f4b5543490a4ef97f69df97f9b165f76402213c692a55be",
"md5": "fc04d168752ea2620e7d884c26e5588a",
"sha256": "abd8d72fc3f76bf12d087255ea3905a5ca5e948a523411df127b812a2d34212c"
},
"downloads": -1,
"filename": "laygo-0.1.2.tar.gz",
"has_sig": false,
"md5_digest": "fc04d168752ea2620e7d884c26e5588a",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 44759,
"upload_time": "2025-07-16T19:41:27",
"upload_time_iso_8601": "2025-07-16T19:41:27.355089Z",
"url": "https://files.pythonhosted.org/packages/d7/1a/1832504b8a826f4b5543490a4ef97f69df97f9b165f76402213c692a55be/laygo-0.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-16 19:41:27",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "ringoldsdev",
"github_project": "laygo-python",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "laygo"
}
