Examples

This section contains practical examples demonstrating Flokkit Flow’s capabilities.

Basic Examples

• Basic Flow - Simple pipelines and transformations

• Backpressure - Handling fast producers and slow consumers

• Filtering and Tapping - Data filtering and observation

Advanced Examples

• Async I/O - Working with async functions and I/O

• Lifecycle Management - Resource management with hooks

• Error Handling - Robust error handling patterns

Pattern Examples

• ETL Pipeline - Extract, Transform, Load pattern

• Stream Processing - Real-time data streams

• Batch Processing - Efficient batch operations

Running Examples

All examples can be found in the examples/ directory:

# List all examples
just example

# Run a specific example
just example 1
just example 01_basic_flow

# Run all examples
just examples

Example Files

1. 01_basic_flow.py - Introduction to Flokkit Flow’s fluent API

2. 02_backpressure.py - Natural backpressure with bounded queues

3. 03_queue_strategies.py - Different strategies for full queues

4. 04_split_merge.py - Parallel processing with fan-out/fan-in

5. 05_tap_filter.py - Observing and filtering data streams

6. 06_caching.py - Memoization for expensive operations

7. 07_async_sources.py - Async data sources and sinks

8. 08_blocking_io.py - Handling blocking I/O operations

9. 09_pipeline_pattern.py - ETL pipeline with auto-completion

10. 10_server_pattern.py - Continuous processing server

11. 11_lifecycle_hooks.py - Resource management with lifecycle hooks

12. 12_dynamic_routing.py - Content-based routing

Contributing Examples

We welcome example contributions! When adding examples:

1. Use descriptive names (e.g., 13_redis_pubsub.py)

2. Include a docstring explaining what the example demonstrates

3. Add comments explaining key concepts

4. Keep examples focused on one concept

5. Update this documentation

Example Template

"""
Example Name

Demonstrates:
- Key concept 1
- Key concept 2
- Key concept 3
"""

import asyncio
from flow import flow

async def main():
    # Create flow
    builder = flow("Example Flow")
    
    # Define components
    def source():
        """Generate example data."""
        yield from range(5)
    
    def process(x: int) -> int:
        """Process data."""
        return x * 2
    
    def sink(x: int) -> None:
        """Output results."""
        print(f"Result: {x}")
    
    # Build pipeline
    (builder
        .source(source)
        .to(process)
        .to(sink))
    
    # Run
    await builder.run()
    
    print("Example completed!")

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