### Pipeline Example - Python

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/getting-started/quickstart.md

Illustrates creating multi-stage processing pipelines with AntFlow. This example defines separate stages for fetching and processing data, showcasing how to sequence asynchronous tasks. It requires the 'antflow' library and 'asyncio'.

```python
import asyncio
from antflow import Pipeline, Stage

async def fetch(x):
    return f"data_{x}"

async def process(x):
    return x.upper()

async def main():
    # Define stages
    stage1 = Stage(name="Fetch", workers=3, tasks=[fetch])
    stage2 = Stage(name="Process", workers=2, tasks=[process])

    # Create and run pipeline
    pipeline = Pipeline(stages=[stage1, stage2])
    results = await pipeline.run(range(10))
    print(results)

asyncio.run(main())
```

--------------------------------

### Progress Visualization with Pipeline.quick() Examples

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/progress.md

Provides examples of integrating progress visualization (`progress=True` and `dashboard`) when using the `Pipeline.quick()` method.

```python
# With quick()
results = await Pipeline.quick(items, process, workers=10, progress=True)

# With dashboard
results = await Pipeline.quick(
    items,
    [fetch, process, save],
    workers=5,
    dashboard="compact"
)
```

--------------------------------

### Install Dependencies with Pip

Source: https://github.com/rodolfonobrega/antflow/blob/main/examples/web_dashboard/README.md

Installs the necessary libraries (FastAPI and Uvicorn) for running the AntFlow web dashboard. This command should be executed in your terminal.

```bash
pip install fastapi uvicorn
```

--------------------------------

### Start AntFlow Web Dashboard Server

Source: https://github.com/rodolfonobrega/antflow/blob/main/examples/web_dashboard/README.md

Commands to start the AntFlow web dashboard server. You can either run it directly using Python or with Uvicorn for development, and then access it via your web browser.

```bash
# From the web_dashboard directory
cd examples/web_dashboard

# Start the server
python server.py
# or
uvicorn server:app --reload

# Open in browser
open http://localhost:8000
```

--------------------------------

### Install Antflow Documentation Dependencies

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/getting-started/installation.md

Installs optional dependencies required to build the Antflow documentation locally. This includes MkDocs and related tools for API documentation generation.

```bash
pip install -e ".[docs]"
```

--------------------------------

### Logging Dashboard Example (Python)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/custom-dashboard.md

A simple custom dashboard implementation in Python that logs pipeline start, progress, and completion status to the console. It utilizes the DashboardProtocol.

```python
from antflow import Pipeline, Stage

class LoggingDashboard:
    def on_start(self, pipeline, total_items):
        print(f"Starting pipeline with {total_items} items")

    def on_update(self, snapshot):
        stats = snapshot.pipeline_stats
        print(f"Progress: {stats.items_processed}/{stats.items_processed + stats.items_in_flight}")

    def on_finish(self, results, summary):
        print(f"Done! {len(results)} succeeded, {summary.total_failed} failed")

# Use it
pipeline = Pipeline(stages=[...])
results = await pipeline.run(items, custom_dashboard=LoggingDashboard())
```

--------------------------------

### AntFlow Quick Start Example (Python)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/index.md

A simple, runnable example demonstrating the basic usage of AntFlow for asynchronous task processing. It defines two pipeline stages: `fetch_data` and `process_data`, each with a configurable number of workers. The example shows how to create a `Pipeline`, run it with a list of items, and process the results.

```python
import asyncio
from antflow import Pipeline, Stage

async def fetch_data(item_id):
    # Simulate fetching data asynchronously
    await asyncio.sleep(0.05)
    return f"data_{item_id}"

async def process_data(data):
    # Simulate processing data asynchronously
    await asyncio.sleep(0.02)
    return data.upper()

async def main():
    # Define pipeline stages with specific worker counts
    fetch_stage = Stage(name="Fetch", workers=10, tasks=[fetch_data])
    process_stage = Stage(name="Process", workers=5, tasks=[process_data])

    # Create a pipeline from the defined stages
    pipeline = Pipeline(stages=[fetch_stage, process_stage])

    # Run the pipeline for 100 items and print the number of results
    results = await pipeline.run(range(100))
    print(f"Processed {len(results)} items")

# Execute the main asynchronous function
asyncio.run(main())

```

--------------------------------

### Rich Dashboard Subclass Example (Python)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/custom-dashboard.md

An example of creating a more advanced dashboard by subclassing `BaseDashboard` in Python. This example displays progress percentage dynamically using rich text.

```python
from antflow.display import BaseDashboard

class MyRichDashboard(BaseDashboard):
    def __init__(self):
        self.total = 0

    def on_start(self, pipeline, total_items):
        self.total = total_items
        print("Starting...")

    def render(self, snapshot):
        # This is called by on_update
        stats = snapshot.pipeline_stats
        pct = (stats.items_processed / self.total * 100) if self.total else 0
        print(f"\rProgress: {pct:.1f}%", end="")

    def on_finish(self, results, summary):
        print(f"\nDone: {len(results)} results")
```

--------------------------------

### Progress Visualization with Builder API Examples

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/progress.md

Shows how to implement progress visualization options (`dashboard`) when constructing pipelines using the AntFlow Builder API.

```python
results = await (
    Pipeline.create()
    .add("Fetch", fetch, workers=10)
    .add("Process", process, workers=5)
    .run(items, dashboard="detailed")
)
```

--------------------------------

### Install Antflow from Source using pip

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/getting-started/installation.md

Installs Antflow directly from its GitHub repository. This method is useful for development or when needing the latest unreleased features. It requires Git to be installed.

```bash
git clone https://github.com/rodolfonobrega/antflow.git
cd antflow
pip install -e ".[dev]"
```

--------------------------------

### Verify Antflow Installation

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/getting-started/installation.md

Confirms that Antflow has been successfully installed by importing the library and printing its version. This Python script should run without errors if the installation was successful.

```python
import antflow
print(antflow.__version__)
```

--------------------------------

### AsyncExecutor Example - Python

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/getting-started/quickstart.md

Demonstrates how to use the AsyncExecutor for concurrent asynchronous operations. It mimics the concurrent.futures API, allowing for easy mapping of async functions over iterables. Requires the 'antflow' library and 'asyncio'.

```python
import asyncio
from antflow import AsyncExecutor

async def process_item(x):
    await asyncio.sleep(0.1)
    return x * 2

async def main():
    async with AsyncExecutor(max_workers=5) as executor:
        # Map over items - returns list directly
        results = await executor.map(process_item, range(10))
        print(results)  # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

asyncio.run(main())
```

--------------------------------

### Execute AntFlow Basic Pipeline Example

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/contributing.md

Runs the basic pipeline example script from the `examples/` directory. This demonstrates the creation and execution of a simple pipeline in AntFlow.

```python
python examples/basic_pipeline.py
```

--------------------------------

### Install Antflow Development Dependencies

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/getting-started/installation.md

Installs optional development dependencies for Antflow, including testing and linting tools. This command should be run from the root of the cloned Antflow repository.

```bash
pip install -e ".[dev]"
```

--------------------------------

### Use Detailed Dashboard with Pipeline.quick()

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/progress.md

Shows how to enable a 'detailed' dashboard for more in-depth pipeline monitoring using `Pipeline.quick()`.

```python
import asyncio
from antflow import Pipeline

async def task(x):
    await asyncio.sleep(0.01)
    return x * 2

async def main():
    items = range(100)
    # Use: "compact", "detailed", or "full"
    results = await Pipeline.quick(items, task, workers=5, dashboard="detailed")

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Complete Antflow Basic Example with AsyncExecutor

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/examples/basic.md

A comprehensive example showcasing Antflow's AsyncExecutor for parallel processing and its Pipeline for multi-stage data processing. It demonstrates creating stages, running tasks, and printing results for both components.

```python
import asyncio
from antflow import AsyncExecutor, Pipeline, Stage

# Executor example
async def double(x):
    await asyncio.sleep(0.1)
    return x * 2

# Pipeline example
async def fetch(x):
    return f"item_{x}"

async def process(x):
    return x.upper()

async def main():
    # Use executor for quick parallel processing
    print("=== AsyncExecutor ===")
    async with AsyncExecutor(max_workers=5) as executor:
        results = await executor.map(double, range(5))
        print(f"Executor results: {results}")

    # Use pipeline for multi-stage processing
    print("\n=== Pipeline ===")
    stage1 = Stage(name="Fetch", workers=3, tasks=[fetch])
    stage2 = Stage(name="Process", workers=2, tasks=[process])

    pipeline = Pipeline(stages=[stage1, stage2])
    results = await pipeline.run(range(5))

    print(f"Pipeline results:")
    for r in results:
        print(f"  {r['id']}: {r['value']}")

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Install Antflow from PyPI using pip

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/getting-started/installation.md

Installs the Antflow library using pip, the Python package installer. Ensure Python 3.9+ and pip are installed. This command fetches and installs the latest stable version of Antflow.

```bash
pip install antflow
```

--------------------------------

### Execute AntFlow Real-World Example

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/contributing.md

Executes a real-world scenario example script from the `examples/` directory. This demonstrates a practical application of AntFlow.

```python
python examples/real_world_example.py
```

--------------------------------

### Execute AntFlow Basic Executor Example

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/contributing.md

Runs the basic executor example script located in the `examples/` directory. This script demonstrates a fundamental use case of the AntFlow executor.

```python
python examples/basic_executor.py
```

--------------------------------

### Execute AntFlow Advanced Pipeline Example

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/contributing.md

Runs an advanced pipeline example script found in the `examples/` directory. This showcases more complex pipeline configurations and features.

```python
python examples/advanced_pipeline.py
```

--------------------------------

### Build and Serve AntFlow Documentation

Source: https://github.com/rodolfonobrega/antflow/blob/main/CONTRIBUTING.md

Installs documentation-specific dependencies and then serves the documentation locally using MkDocs. This allows for previewing documentation changes before building for production.

```bash
pip install -e ".[docs]"
mkdocs serve
mkdocs build
```

--------------------------------

### Check Pipeline Status with httpx

Source: https://github.com/rodolfonobrega/antflow/blob/main/examples/web_dashboard/README.md

An example of how to integrate with the AntFlow web dashboard from another Python application by making an asynchronous GET request to the `/api/status` endpoint using the `httpx` library.

```python
import httpx

async def check_pipeline_status():
    async with httpx.AsyncClient() as client:
        response = await client.get("http://localhost:8000/api/status")
        return response.json()
```

--------------------------------

### Complete ETL Pipeline Example in Python

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/pipeline.md

This comprehensive Python script outlines a complete Extract, Transform, Load (ETL) pipeline using AntFlow. It defines an ETLProcessor class with methods for each stage (extract, validate, transform, enrich, load) and configures a Pipeline with multiple Stages, each having specific worker counts, retry strategies, and task attempts. Dependencies include 'asyncio' and 'antflow'.

```python
import asyncio
from antflow import Pipeline, Stage

class ETLProcessor:
    async def extract(self, item_id):
        # Fetch from API/database
        await asyncio.sleep(0.1)
        return {"id": item_id, "data": f"raw_{item_id}"}

    async def validate(self, data):
        # Validate data
        if "data" not in data:
            raise ValueError("Invalid data")
        return data

    async def transform(self, data):
        # Transform data
        data["processed"] = data["data"].upper()
        return data

    async def enrich(self, data):
        # Enrich with additional data
        data["metadata"] = {"timestamp": "2025-10-09"}
        return data

    async def load(self, data):
        # Save to database
        await asyncio.sleep(0.1)
        data["saved"] = True
        return data

async def main():
    processor = ETLProcessor()

    # Extract stage with high concurrency
    extract_stage = Stage(
        name="Extract",
        workers=10,
        tasks=[processor.extract],
        retry="per_task",
        task_attempts=5,
        task_wait_seconds=2.0
    )

    # Transform stage with validation
    transform_stage = Stage(
        name="Transform",
        workers=5,
        tasks=[processor.validate, processor.transform, processor.enrich],
        retry="per_stage",  # Transactional
        stage_attempts=3
    )

    # Load stage with retries
    load_stage = Stage(
        name="Load",
        workers=3,
        tasks=[processor.load],
        retry="per_task",
        task_attempts=5,
        task_wait_seconds=3.0
    )

    # Build pipeline
    pipeline = Pipeline(
        stages=[extract_stage, transform_stage, load_stage]
    )

    # Process items
    item_ids = range(100)
    results = await pipeline.run(item_ids)

    print(f"Processed {len(results)} items")
    print(f"Stats: {pipeline.get_stats()}")

asyncio.run(main())
```

--------------------------------

### Google-Style Docstring Example

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/contributing.md

Illustrates the Google-style docstring format recommended for AntFlow. It includes sections for a brief description, arguments, return values, and raised exceptions.

```python
async def my_function(x: int, y: str = "default") -> bool:
    """
    Brief description of what the function does.

    Args:
        x: Description of x parameter
        y: Description of y parameter with default

    Returns:
        Description of return value

    Raises:
        ValueError: Description of when this is raised
    """
    pass
```

--------------------------------

### AntFlow Installation (Bash)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/index.md

Provides the command to install the AntFlow library using pip. This is the standard method for adding Python packages to a project's environment, enabling the use of AntFlow's asynchronous execution capabilities.

```bash
pip install antflow

```

--------------------------------

### Python Pipeline Usage Example

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/api/pipeline.md

Demonstrates how to define Stages with tasks and workers, then orchestrate them using a Pipeline to process a list of items asynchronously. This example requires the antflow library and asyncio.

```python
import asyncio
from antflow import Pipeline, Stage

async def fetch_data(url):
    # ... fetch logic ...
    return data

async def process_data(data):
    # ... process logic ...
    return result

async def main():
    # Define stages
    stage1 = Stage(name="Fetch", workers=5, tasks=[fetch_data])
    stage2 = Stage(name="Process", workers=2, tasks=[process_data])

    # Create pipeline
    pipeline = Pipeline(stages=[stage1, stage2])

    # Run
    urls = ["http://example.com/1", "http://example.com/2"]
    results = await pipeline.run(urls)

    for result in results:
        print(f"ID: {result.id}, Value: {result.value}")

```

--------------------------------

### AntFlow AsyncExecutor: Compare map() vs as_completed() (Python)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/examples/basic.md

Highlights the fundamental difference between AsyncExecutor.map() (maintains input order) and AsyncExecutor.as_completed() (processes in completion order). This example uses tasks with variable delays to clearly illustrate the ordering behavior. Requires 'antflow' and 'asyncio'.

```python
import asyncio
from antflow import AsyncExecutor

async def variable_delay_task(x: int) -> str:
    # Item 0 is slow, others are fast
    delay = 2.0 if x == 0 else 0.3
    await asyncio.sleep(delay)
    return f"Result-{x}"

async def main():
    async with AsyncExecutor(max_workers=5) as executor:
        items = range(5)

        # map(): Results in INPUT order
        print("=== map() - Input Order ===")
        results = await executor.map(variable_delay_task, items)
        for r in results:
            print(f"  {r}")

        # as_completed(): Results in COMPLETION order
        print("\n=== as_completed() - Completion Order ===")
        futures = [executor.submit(variable_delay_task, i) for i in items]
        async for future in executor.as_completed(futures):
            print(f"  {await future.result()}")

asyncio.run(main())
```

--------------------------------

### WebSocket API

Source: https://github.com/rodolfonobrega/antflow/blob/main/examples/web_dashboard/README.md

Provides real-time pipeline status updates via WebSocket connections.

```APIDOC
## WebSocket /ws

### Description
Establishes a WebSocket connection for receiving real-time updates on the pipeline status. Data is sent as JSON objects.

### Endpoint
/ws

### Parameters
None

### Connection Example
```javascript
const socket = new WebSocket("ws://localhost:8000/ws");

socket.onmessage = function(event) {
    const data = JSON.parse(event.data);
    console.log("Pipeline Update:", data);
    // Update UI with new status
};

socket.onopen = function(event) {
    console.log("WebSocket connection opened");
};

socket.onclose = function(event) {
    console.log("WebSocket connection closed");
};

socket.onerror = function(event) {
    console.error("WebSocket error observed:", event);
};
```

### Message Format
Messages sent over the WebSocket are in the same JSON format as the `/api/status` endpoint response.

#### Response Example (per message)
```json
{
  "is_running": true,
  "progress": {
    "processed": 47,
    "failed": 5,
    "in_flight": 12,
    "total": 100
  },
  "stages": {
    "Fetch": {"pending": 8, "active": 5, "completed": 87, "failed": 0},
    "Process": {"pending": 35, "active": 3, "completed": 54, "failed": 2},
    "Save": {"pending": 20, "active": 2, "completed": 30, "failed": 3}
  },
  "workers": [
    {"name": "Fetch-W0", "stage": "Fetch", "status": "busy", "current_item": 42},
    {"name": "Fetch-W1", "stage": "Fetch", "status": "idle", "current_item": null}
  ]
}
```
```

--------------------------------

### Import Organization Example

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/contributing.md

Shows the recommended order for organizing imports in AntFlow: standard library, third-party, and local application imports.

```python
import asyncio
import logging
from typing import Any, List

from tenacity import retry

from antflow.exceptions import AntFlowError
from antflow.types import TaskFunc
```

--------------------------------

### Define AntFlow Pipeline Structure

Source: https://github.com/rodolfonobrega/antflow/blob/main/examples/web_dashboard/README.md

Demonstrates how to define the structure of an AntFlow pipeline by specifying stages, their worker counts, and the tasks associated with each stage.

```python
pipeline = Pipeline(
    stages=[
        Stage("Extract", workers=10, tasks=[extract_task]),
        Stage("Transform", workers=5, tasks=[transform_task]),
        Stage("Load", workers=3, tasks=[load_task]),
    ]
)
```

--------------------------------

### Serve Antflow Documentation Locally with MkDocs

Source: https://github.com/rodolfonobrega/antflow/blob/main/README.md

Instructions to build and serve the Antflow project documentation locally. This requires Python 3.9+ and the mkdocs-material package. After installation, the documentation can be accessed via a local web server.

```bash
pip install mkdocs-material
python -m mkdocs serve
```

--------------------------------

### Basic Dashboard Usage with AntFlow (Python)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/dashboard.md

Demonstrates how to use AntFlow's `Pipeline.quick` and `Pipeline.create().add().run()` methods to display progress bars and dashboards during pipeline execution. It shows examples for simple progress, compact dashboards, and detailed dashboards suitable for multi-stage pipelines.

```python
import asyncio
from antflow import Pipeline

async def task(x):
    await asyncio.sleep(0.1)
    return x * 2

async def main():
    items = range(100)
    
    # 1. Simple progress bar
    await Pipeline.quick(items, task, workers=10, progress=True)

    # 2. Compact dashboard
    await Pipeline.quick(items, task, workers=10, dashboard="compact")

    # 3. Detailed dashboard (Recommended for multi-stage)
    await (
        Pipeline.create()
        .add("Fetch", task, workers=10)
        .add("Process", task, workers=5)
        .run(items, dashboard="detailed")
    )

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### FastAPI WebSocket with Custom Dashboard for Push Updates

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/custom-dashboard.md

This example demonstrates using a custom dashboard class with FastAPI WebSockets for push-based pipeline updates. The `WebSocketDashboard` class implements `on_start`, `on_update`, and `on_finish` methods to send pipeline status events (start, update, finish) to the WebSocket client. This avoids polling and provides immediate feedback.

```python
import asyncio
from fastapi import FastAPI, WebSocket
from antflow import Pipeline, Stage

app = FastAPI()

class WebSocketDashboard:
    def __init__(self, websocket):
        self.ws = websocket

    def on_start(self, pipeline, total_items):
        asyncio.create_task(self.ws.send_json({
            "type": "start",
            "total": total_items,
            "stages": [s.name for s in pipeline.stages]
        }))

    def on_update(self, snapshot):
        stats = snapshot.pipeline_stats
        asyncio.create_task(self.ws.send_json({
            "type": "update",
            "processed": stats.items_processed,
            "failed": stats.items_failed,
            "in_flight": stats.items_in_flight,
            "stages": {
                name: {"pending": s.pending_items, "active": s.in_progress_items, "done": s.completed_items}
                for name, s in stats.stage_stats.items()
            }
        }))

    def on_finish(self, results, summary):
        asyncio.create_task(self.ws.send_json({
            "type": "finish",
            "results": len(results),
            "failed": summary.total_failed,
            "errors": summary.errors_by_type
        }))

# Usage with FastAPI
@app.websocket("/ws/pipeline/run")
async def run_with_websocket(websocket: WebSocket, items: list):
    await websocket.accept()

    pipeline = Pipeline(stages=[...]) # Define your stages here
    dashboard = WebSocketDashboard(websocket)

    results = await pipeline.run(items, custom_dashboard=dashboard)
    # Dashboard automatically sends start, updates, and finish events

```

--------------------------------

### DashboardProtocol Interface (Python)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/custom-dashboard.md

Defines the structure of a custom dashboard by implementing the DashboardProtocol. It includes methods for pipeline start, periodic updates, and completion.

```python
from antflow import DashboardProtocol

class MyDashboard:
    def on_start(self, pipeline, total_items):
        """Called when pipeline execution starts."""
        pass

    def on_update(self, snapshot):
        """Called periodically with current pipeline state."""
        pass

    def on_finish(self, results, summary):
        """Called when pipeline execution completes."""
        pass
```

--------------------------------

### Custom Dashboard Implementation with DashboardProtocol (Python)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/dashboard.md

Demonstrates how to create a custom dashboard for AntFlow by implementing the `DashboardProtocol`. The `MyLogDashboard` class shows example methods like `on_start`, `on_update`, and `on_finish` that can be customized to provide specific display logic when passed to `pipeline.run()`.

```python
from antflow import DashboardProtocol

class MyLogDashboard:
    def on_start(self, pipeline, total_items):
        print(f"Starting {total_items} items")

    def on_update(self, snapshot):
        print(f"Progress: {snapshot.pipeline_stats.items_processed}")

    def on_finish(self, results, summary):
        print("Done!")
```

--------------------------------

### Enable Minimal Progress Bar with Pipeline.quick()

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/progress.md

Demonstrates adding a simple, zero-configuration progress bar to a pipeline using the `progress=True` flag with `Pipeline.quick()`.

```python
import asyncio
from antflow import Pipeline

async def task(x):
    return x * 2

async def main():
    results = await Pipeline.quick(range(100), task, workers=10, progress=True)

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Programmatically Get Pipeline Summary Statistics (Python)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/pipeline.md

Retrieves a summary of pipeline statistics after execution using `pipeline.get_stats()`. This provides high-level metrics and detailed per-stage statistics.

```python
stats = pipeline.get_stats()

print(f"Total processed: {stats.items_processed}")
print(f"Total failed: {stats.items_failed}")

# Access per-stage metrics
for stage_name, stage_stat in stats.stage_stats.items():
    print(f"Stage {stage_name}:")
    print(f"  Completed: {stage_stat.completed_items}")
    print(f"  Failed: {stage_stat.failed_items}")
    print(f"  In progress: {stage_stat.in_progress_items}")
```

--------------------------------

### Getting Antflow Pipeline Statistics

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/examples/basic.md

Shows how to retrieve statistics from an Antflow pipeline after execution. It prints the counts of items processed, failed, and in-flight, along with queue sizes for each stage.

```python
import asyncio
from antflow import Pipeline, Stage

async def main():
    pipeline = Pipeline(stages=[extract_stage, transform_stage])
    results = await pipeline.run(range(100))

    # Get statistics
    stats = pipeline.get_stats()
    print(f"Items processed: {stats.items_processed}")
    print(f"Items failed: {stats.items_failed}")
    print(f"Items in-flight: {stats.items_in_flight}")
    print(f"Queue sizes: {stats.queue_sizes}")

asyncio.run(main())
```

--------------------------------

### Python: Task-Level Callbacks for Granular Monitoring

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/custom-dashboard.md

Configures `StatusTracker` to capture fine-grained events at the task level. This includes callbacks for task start, completion, retry, and failure, enabling detailed debugging and performance analysis.

```python
tracker = StatusTracker(
    on_status_change=handle_item_status,    # Item-level events
    on_task_start=handle_task_start,        # Task started
    on_task_complete=handle_task_complete,  # Task succeeded
    on_task_retry=handle_task_retry,        # Task being retried
    on_task_fail=handle_task_fail,          # Task failed
)
```

--------------------------------

### FastAPI REST Endpoint to Start AntFlow Pipeline

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/custom-dashboard.md

A FastAPI endpoint to initiate an AntFlow pipeline. It accepts a list of items to process and configures a simple pipeline with one stage. The pipeline runs asynchronously in the background.

```python
from fastapi import FastAPI
from antflow import Pipeline, Stage
import asyncio

app = FastAPI()
pipeline = None  # Will be set when pipeline starts
total_items = 0

# Assume my_task is defined elsewhere
def my_task(item):
    # Placeholder for actual task logic
    pass

@app.post("/api/pipeline/start")
async def start_pipeline(items: list):
    """Start pipeline processing."""
    global pipeline, total_items
    total_items = len(items)

    pipeline = Pipeline(stages=[
        Stage("Process", workers=5, tasks=[my_task])
    ])

    # Run in background
    asyncio.create_task(pipeline.run(items))
    return {"status": "started", "total": total_items}
```

--------------------------------

### Initialize FullDashboard (Python)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/api/display.md

Initializes the FullDashboard, the most comprehensive monitoring option, providing an overview, stage metrics, worker monitoring, and item tracking. It allows configuration of refresh rate and the maximum number of items shown. Requires a StatusTracker for item tracking.

```python
from antflow.display import FullDashboard
from antflow import StatusTracker, Pipeline

# Constructor usage
tracker = StatusTracker()
pipeline = Pipeline(stages=[...], status_tracker=tracker)

dashboard = FullDashboard(
    refresh_rate=4.0,
    max_items_shown=20,
)
```

--------------------------------

### Querying Item Status and History

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/pipeline.md

Provides examples for querying the status and history of specific items within the pipeline using the `StatusTracker`. It covers retrieving the current status, listing all failed items, and accessing the historical events for a given item.

```python
# Get specific item status
status = tracker.get_status(item_id=42)
print(f"Item 42: {status.status} @ {status.stage}")

# Get all failed items
failed = tracker.get_by_status("failed")
for event in failed:
    print(f"Item {event.item_id}: {event.metadata['error']}")

# Get item history
history = tracker.get_history(item_id=42)
for event in history:
    print(f"{event.timestamp}: {event.status}")
```

--------------------------------

### Enable Full Dashboard with Item Tracking using StatusTracker

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/progress.md

Illustrates enabling the 'full' dashboard and incorporating `StatusTracker` for individual item tracking within an AntFlow pipeline.

```python
import asyncio
from antflow import Pipeline, StatusTracker

async def task(x):
    return x

async def main():
    tracker = StatusTracker()
    results = await (
        Pipeline.create()
        .add("Process", task, workers=5)
        .with_tracker(tracker)
        .run(range(50), dashboard="full")
    )

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Python: Get Per-Stage Pipeline Statistics

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/custom-dashboard.md

Retrieves and prints detailed statistics for each stage within a pipeline using the `stage_stats` attribute of `pipeline_stats`. This method is useful for obtaining a granular overview of item progression through different pipeline stages.

```python
class MultiStageDashboard:
    def on_update(self, snapshot):
        stats = snapshot.pipeline_stats

        print(f"Overall: {stats.items_processed}/{self.total} completed end-to-end")
        print()

        # Per-stage breakdown
        for stage_name, stage_stat in stats.stage_stats.items():
            total_in_stage = (
                stage_stat.pending_items +
                stage_stat.in_progress_items +
                stage_stat.completed_items +
                stage_stat.failed_items
            )
            print(f"{stage_name}:")
            print(f"  Pending:     {stage_stat.pending_items}")
            print(f"  In Progress: {stage_stat.in_progress_items}")
            print(f"  Completed:   {stage_stat.completed_items}")
            print(f"  Failed:      {stage_stat.failed_items}")
```

--------------------------------

### Get Antflow Pipeline Worker Names

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/worker-tracking.md

Retrieve a dictionary of all worker names within an Antflow pipeline, categorized by stage. This is useful for pre-run setup, monitoring dashboard configuration, or understanding pipeline structure.

```python
from antflow import Pipeline, Stage

stage1 = Stage(name="Fetch", workers=3, tasks=[fetch_data])
stage2 = Stage(name="Process", workers=5, tasks=[process_data])

pipeline = Pipeline(stages=[stage1, stage2])

worker_names = pipeline.get_worker_names()
print(worker_names)
# {
#     "Fetch": ["Fetch-W0", "Fetch-W1", "Fetch-W2"],
#     "Process": ["Process-W0", "Process-W1", "Process-W2", "Process-W3", "Process-W4"]
# }
```

--------------------------------

### Implement Custom Dashboards with DashboardProtocol in Python

Source: https://context7.com/rodolfonobrega/antflow/llms.txt

Demonstrates creating a custom dashboard by implementing the DashboardProtocol. This involves defining `on_start`, `on_update`, and `on_finish` methods to integrate with external systems. The example shows progress tracking and error reporting during pipeline execution.

```python
import asyncio
import sys
from datetime import datetime
from antflow import Pipeline, Stage, DashboardSnapshot, PipelineResult, ErrorSummary

class SimpleDashboard:
    """Custom dashboard implementing DashboardProtocol."""

    def on_start(self, pipeline, total_items: int) -> None:
        print(f"[{datetime.now().strftime('%H:%M:%S')}] Starting with {total_items} items")
        self.total = total_items
        self.stages = [s.name for s in pipeline.stages]

    def on_update(self, snapshot: DashboardSnapshot) -> None:
        stats = snapshot.pipeline_stats
        pct = (stats.items_processed / self.total * 100) if self.total else 0

        # Per-stage progress
        stage_info = " | ".join(
            f"{name}: {s.completed_items}/{s.pending_items + s.in_progress_items + s.completed_items}"
            for name, s in stats.stage_stats.items()
        )

        sys.stdout.write(
            f"\r[{datetime.now().strftime('%H:%M:%S')}] "
            f"Progress: {pct:5.1f}% | "
            f"Done: {stats.items_processed} | "
            f"Failed: {stats.items_failed} | "
            f"{stage_info}   "
        )
        sys.stdout.flush()

    def on_finish(self, results: list, summary: ErrorSummary) -> None:
        print()
        print(f"[{datetime.now().strftime('%H:%M:%S')}] Complete!")
        print(f"  Results: {len(results)}")
        print(f"  Failed: {summary.total_failed}")
        if summary.errors_by_type:
            print(f"  Errors by type: {summary.errors_by_type}")

async def fetch(x: int) -> dict:
    await asyncio.sleep(0.1)
    return {"id": x}

async def process(item: dict) -> dict:
    import random
    await asyncio.sleep(0.05)
    if random.random() < 0.1:
        raise ValueError("Random error")
    item["done"] = True
    return item

async def main():
    pipeline = Pipeline(
        stages=[
            Stage(name="Fetch", workers=5, tasks=[fetch], task_attempts=2),
            Stage(name="Process", workers=3, tasks=[process], task_attempts=2),
        ]
    )

    results = await pipeline.run(
        list(range(30)),
        custom_dashboard=SimpleDashboard()
    )

asyncio.run(main())

```

--------------------------------

### Python: Manual Pipeline Polling without DashboardProtocol

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/custom-dashboard.md

Provides a method for manually polling pipeline status without relying on the `DashboardProtocol`. This involves starting the pipeline in the background, feeding items, and then entering a loop to periodically fetch `dashboard_snapshot` for status updates.

```python
import asyncio
from antflow import Pipeline, Stage

async def main():
    pipeline = Pipeline(stages=[...])

    # Start pipeline in background
    await pipeline.start()
    await pipeline.feed(items)

    # Manual polling loop
    while True:
        snapshot = pipeline.get_dashboard_snapshot()
        stats = snapshot.pipeline_stats

        print(f"\rProcessed: {stats.items_processed}", end="")

        if stats.items_processed + stats.items_failed >= len(items):
            break

        await asyncio.sleep(0.1)

    await pipeline.join()
    results = pipeline.results
```

--------------------------------

### Create AsyncExecutor Instance

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/executor.md

Demonstrates how to create an AsyncExecutor instance, specifying the maximum number of worker coroutines. It also shows the recommended usage as an asynchronous context manager.

```python
from antflow import AsyncExecutor

# Create executor with 5 workers
executor = AsyncExecutor(max_workers=5)

# Or use as context manager (recommended)
async with AsyncExecutor(max_workers=5) as executor:
    # Use executor here
    pass
```

--------------------------------

### Configure AsyncExecutor Worker Count for Performance in Python

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/executor.md

Provides guidance on choosing the appropriate number of workers for Antflow's AsyncExecutor based on task type (I/O-bound vs. CPU-bound), rate limits, and memory constraints. It includes example configurations for different scenarios.

```python
from antflow import AsyncExecutor

# For I/O-bound tasks (API calls, database queries)
executor = AsyncExecutor(max_workers=50)

# For rate-limited APIs (e.g., 10 requests/second)
executor = AsyncExecutor(max_workers=10)

# For memory-intensive tasks
executor = AsyncExecutor(max_workers=5)
```

--------------------------------

### Python: Combine Dashboard and StatusTracker for Monitoring

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/custom-dashboard.md

Demonstrates how to use both the `DashboardProtocol` (implicitly via `pipeline.run(dashboard=...)`) and `StatusTracker` concurrently. This approach allows for both a visual overview and detailed, event-driven logging, such as logging item failures with metadata.

```python
async def log_failures(event: StatusEvent):
    if event.status == "failed":
        logging.error(f"Item {event.item_id} failed at {event.stage}: {event.metadata}")

tracker = StatusTracker(on_status_change=log_failures)

pipeline = Pipeline(stages=[...], status_tracker=tracker)

# Dashboard for UI + callbacks for logging
results = await pipeline.run(items, dashboard="compact")
```

--------------------------------

### Feed Items with Priorities in AntFlow

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/pipeline.md

This Python example shows how to feed items into an AntFlow pipeline with different priority levels. Lower numbers signify higher priority. The code demonstrates feeding expedited, normal, and low-priority items, illustrating AntFlow's priority queue mechanism. Dependencies include the 'antflow' library.

```python
# Expedited items (Priority 10)
await pipeline.feed(vip_items, priority=10)

# Normal items (Priority 100)
await pipeline.feed(regular_items)

# Background/Low priority (Priority 500)
await pipeline.feed(maintenance_items, priority=500)
```

--------------------------------

### Initialize CompactDashboard (Python)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/api/display.md

Initializes a CompactDashboard, a rich-based dashboard for displaying essential pipeline metrics in a compact format. It can be configured with a refresh rate. This dashboard is suitable for a quick overview of the pipeline's status.

```python
from antflow.display import CompactDashboard

# Constructor usage
dashboard = CompactDashboard(refresh_rate=4.0)
```

--------------------------------

### Flask API for Synchronous Pipeline Status

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/custom-dashboard.md

This Flask example provides a synchronous API endpoint to retrieve the current status of an Antflow pipeline. It defines a `/api/status` route that, if a pipeline is running, fetches and returns the number of processed and failed items, along with active and completed counts for each stage. If no pipeline is running, it returns a `not_running` status.

```python
from flask import Flask, jsonify
import asyncio
from antflow import Pipeline, Stage

app = Flask(__name__)
pipeline = None # Assume pipeline is initialized elsewhere
loop = asyncio.new_event_loop()

@app.route('/api/status')
def get_status():
    if not pipeline:
        return jsonify({"status": "not_running"})

    snapshot = pipeline.get_dashboard_snapshot()
    stats = snapshot.pipeline_stats

    return jsonify({
        "processed": stats.items_processed,
        "failed": stats.items_failed,
        "stages": {
            name: {"done": s.completed_items, "active": s.in_progress_items}
            for name, s in stats.stage_stats.items()
        }
    })

```

--------------------------------

### Complete Antflow Pipeline Example with Worker Activity Tracking (Python)

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/worker-tracking.md

Demonstrates a full Antflow pipeline execution, including defining stages, running items, and tracking worker activity. The `on_status_change` function logs when a worker starts processing an item and when it completes. It also summarizes the total items processed by each worker at the end.

```python
import asyncio
from collections import defaultdict
from antflow import Pipeline, Stage, StatusTracker

async def process_batch(batch_data):
    await asyncio.sleep(0.2)
    return f"processed_{batch_data}"

async def main():
    item_to_worker = {}
    worker_activity = defaultdict(list)

    async def on_status_change(event):
        if event.status == "in_progress":
            item_to_worker[event.item_id] = event.worker_id
            worker_activity[event.worker_id].append(event.item_id)
            print(f"[Worker {event.worker_id:2d}] Processing {event.item_id}")
        elif event.status == "completed":
            print(f"[Worker {event.worker_id:2d}] Completed {event.item_id}")

    tracker = StatusTracker(on_status_change=on_status_change)

    stage = Stage(name="ProcessBatch", workers=5, tasks=[process_batch])
    pipeline = Pipeline(stages=[stage], status_tracker=tracker)

    worker_names = pipeline.get_worker_names()
    print(f"Available workers: {worker_names}\n")

    items = [
        {"id": f"batch_{i:04d}", "value": f"data_{i}"}
        for i in range(20)
    ]

    results = await pipeline.run(items)

    print("\n=== Worker Utilization ===")
    for worker_id in sorted(worker_activity.keys()):
        items_processed = worker_activity[worker_id]
        print(f"Worker {worker_id}: {len(items_processed)} items")

asyncio.run(main())
```

--------------------------------

### Create Pipeline using Fluent Builder API

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/pipeline.md

Demonstrates creating a multi-stage pipeline using Antflow's fluent builder API. This approach is recommended for its conciseness and chainable nature, suitable for most multi-stage pipeline needs. It requires the 'antflow' library.

```python
from antflow import Pipeline

async def main():
    results = await (
        Pipeline.create()
        .add("Fetch", fetch_task, workers=10)
        .add("Process", process_task, workers=5)
        .run(items, progress=True)
    )
```

--------------------------------

### Create Pipeline using Quick One-Liner API

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/pipeline.md

Shows how to create pipelines using the `Pipeline.quick()` method for simple scripts or single-stage processing. This API is efficient for straightforward tasks, supporting both single and multiple tasks within one stage. It requires the 'antflow' library.

```python
from antflow import Pipeline

# Single task
results = await Pipeline.quick(items, process_task, workers=5, progress=True)

# Multiple tasks in one stage
results = await Pipeline.quick(items, [fetch, process], workers=10)
```

--------------------------------

### Pipeline.quick(): Simple Pipeline Creation in Python

Source: https://context7.com/rodolfonobrega/antflow/llms.txt

The `Pipeline.quick()` method offers a concise API for creating single-stage or multi-stage pipelines. When provided with a single task function, it creates one stage; with a list of functions, it creates a stage for each. This method is suitable for quick scripts and straightforward data transformations.

```python
import asyncio
from antflow import Pipeline

async def fetch(x: int) -> dict:
    await asyncio.sleep(0.1)
    return {"id": x, "data": f"item_{x}"}

async def process(item: dict) -> dict:
    item["processed"] = True
    return item

async def save(item: dict) -> str:
    return f"saved_{item['id']}"

async def main():
    items = list(range(20))

    # Single task - creates one stage
    results = await Pipeline.quick(items, fetch, workers=5, progress=True)
    print(f"Fetched: {len(results)} items")

    # Multiple tasks - creates one stage per task
    results = await Pipeline.quick(
        items,
        [fetch, process, save],
        workers=5,
        retries=3,
        dashboard="compact"  # Options: "compact", "detailed", "full"
    )
    print(f"Processed: {len(results)} items")
    print(f"Sample result: {results[0].value}")  # "saved_0"

asyncio.run(main())
```

--------------------------------

### Basic Status Tracking with StatusTracker

Source: https://github.com/rodolfonobrega/antflow/blob/main/docs/user-guide/pipeline.md

Demonstrates how to initialize and use StatusTracker to track items flowing through a pipeline. It shows how to run a pipeline and then query the overall statistics like completed and failed items.

```python
from antflow import Pipeline, Stage, StatusTracker

tracker = StatusTracker()

stage = Stage(
    name="ProcessStage",
    workers=3,
    tasks=[my_task]
)

pipeline = Pipeline(stages=[stage], status_tracker=tracker)
results = await pipeline.run(items)

# Query statistics
stats = tracker.get_stats()
print(f"Completed: {stats['completed']}")
print(f"Failed: {stats['failed']}")
```