# Basic Usage

This guide covers the fundamental patterns for building flows with Flokkit Flow.

## Building Your First Pipeline

### Simple Linear Pipeline

```python
from flow import flow
import asyncio

async def main():
    builder = flow("Simple Pipeline")
    
    # Source: Generate data
    def generate_data():
        for i in range(5):
            yield f"Item {i}"
    
    # Transform: Process data
    def process_item(item: str) -> str:
        return item.upper()
    
    # Sink: Output results
    def print_result(item: str) -> None:
        print(f"Processed: {item}")
    
    # Connect nodes
    (builder
        .source(generate_data)
        .to(process_item)
        .to(print_result))
    
    # Run pipeline
    await builder.run()

asyncio.run(main())
```

### Adding Multiple Transforms

```python
(builder
    .source(read_lines)
    .to(strip_whitespace)
    .to(parse_json)
    .to(validate_schema)
    .to(enrich_data)
    .to(save_to_db))
```

## Working with Different Data Types

### Type-Safe Pipelines

```python
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class User:
    id: int
    name: str
    email: str

@dataclass
class EnrichedUser:
    user: User
    score: float
    tags: List[str]

async def create_user_pipeline():
    builder = flow("User Processing")
    
    def fetch_users() -> User:
        # Fetch from database
        yield User(1, "Alice", "alice@example.com")
        yield User(2, "Bob", "bob@example.com")
    
    def calculate_score(user: User) -> float:
        # Complex scoring logic
        return len(user.name) * 10.0
    
    def enrich_user(user: User) -> EnrichedUser:
        score = calculate_score(user)
        tags = ["active", "verified"] if score > 50 else ["new"]
        return EnrichedUser(user, score, tags)
    
    def save_enriched(enriched: EnrichedUser) -> None:
        print(f"Saving {enriched.user.name} with score {enriched.score}")
    
    (builder
        .source(fetch_users)
        .to(enrich_user)
        .to(save_enriched))
    
    await builder.run()
```

## Filtering Data

### Basic Filtering

```python
def is_valid(item: dict) -> bool:
    return item.get("status") == "active"

(builder
    .source(fetch_items)
    .filter(is_valid)  # Only active items pass through
    .to(process_item))
```

### Filtering with Side Effects

```python
def log_invalid(item: dict) -> bool:
    if item.get("status") != "active":
        logger.warning(f"Filtered out inactive item: {item['id']}")
        return False
    return True

(builder
    .source(fetch_items)
    .filter(log_invalid)
    .to(process_item))
```

## Observing Data with Tap

### Logging Pipeline

```python
def log_throughput(item: Any) -> None:
    logger.info(f"Processing item: {item}")

(builder
    .source(generate_data)
    .tap(log_throughput)  # Observe without consuming
    .to(transform_data)
    .tap(lambda x: metrics.increment("items_processed"))
    .to(save_data))
```

### Debugging with Tap

```python
def debug_print(item: Any, label: str = "") -> None:
    print(f"[DEBUG {label}] {item}")

(builder
    .source(complex_source)
    .tap(lambda x: debug_print(x, "after source"))
    .to(transform_1)
    .tap(lambda x: debug_print(x, "after transform1"))
    .to(transform_2)
    .tap(lambda x: debug_print(x, "after transform2"))
    .to(final_sink))
```

## Splitting and Merging Flows

### Fan-Out Pattern

```python
async def fan_out_example():
    builder = flow("Fan-Out")
    
    source = builder.source(generate_data)
    
    # Split to multiple paths
    path1 = source.to(processor_1)
    path2 = source.to(processor_2)
    path3 = source.to(processor_3)
    
    # Each path processes independently
    path1.to(sink_1)
    path2.to(sink_2)
    path3.to(sink_3)
    
    await builder.run()
```

### Fan-In Pattern

```python
async def fan_in_example():
    builder = flow("Fan-In")
    
    # Multiple sources
    source1 = builder.source(api_source_1)
    source2 = builder.source(api_source_2)
    source3 = builder.source(api_source_3)
    
    # Merge into single stream
    merged = source1.merge_with(source2, source3)
    
    # Process merged stream
    merged.to(unified_processor).to(save_results)
    
    await builder.run()
```

## Caching Expensive Operations

### Enable Caching

```python
def expensive_calculation(data: str) -> float:
    # Simulate expensive operation
    time.sleep(1)
    return len(data) * 42.0

(builder
    .source(generate_strings)
    .to(expensive_calculation, cache=True)  # Results are cached
    .to(print_result))
```

### Custom Cache Key

```python
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_api_call(user_id: int) -> dict:
    # This will only be called once per unique user_id
    return fetch_user_data(user_id)

(builder
    .source(get_user_ids)
    .to(cached_api_call)
    .to(process_user))
```

## Error Handling Patterns

### Continue on Error

```python
def safe_process(item: dict) -> Optional[dict]:
    try:
        return risky_transform(item)
    except ValueError as e:
        logger.error(f"Failed to process {item}: {e}")
        return None  # Continue with None

(builder
    .source(fetch_items)
    .to(safe_process)
    .filter(lambda x: x is not None)  # Skip failed items
    .to(save_result))
```

### Error Output Stream

```python
async def error_handling_flow():
    builder = flow("Error Handling")
    
    def process_with_errors(item: str) -> str:
        if "error" in item:
            raise ValueError(f"Invalid item: {item}")
        return item.upper()
    
    # Main path
    source = builder.source(generate_mixed_data)
    
    # Split good and bad items
    good_items = source.filter(lambda x: "error" not in x)
    bad_items = source.filter(lambda x: "error" in x)
    
    # Process good items
    good_items.to(process_with_errors).to(save_good)
    
    # Handle bad items
    bad_items.to(log_error).to(save_to_error_queue)
    
    await builder.run()
```

## Working with External Resources

### Database Pipeline

```python
class DatabasePipeline:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.pool = None
    
    async def setup(self):
        self.pool = await asyncpg.create_pool(self.connection_string)
    
    async def cleanup(self):
        if self.pool:
            await self.pool.close()
    
    async def fetch_records(self):
        async with self.pool.acquire() as conn:
            async for record in conn.cursor("SELECT * FROM users"):
                yield dict(record)
    
    async def update_record(self, record: dict):
        async with self.pool.acquire() as conn:
            await conn.execute(
                "UPDATE users SET processed = true WHERE id = $1",
                record["id"]
            )

async def run_db_pipeline():
    db = DatabasePipeline("postgresql://localhost/mydb")
    await db.setup()
    
    try:
        builder = flow("Database Processing")
        
        (builder
            .source(db.fetch_records)
            .to(transform_record)
            .to(db.update_record))
        
        await builder.run()
    finally:
        await db.cleanup()
```

## Best Practices

### 1. Keep Functions Pure

```python
# Good: Pure function
def double(x: int) -> int:
    return x * 2

# Bad: Side effects in transform
def double_with_side_effect(x: int) -> int:
    print(f"Doubling {x}")  # Use tap() instead
    return x * 2
```

### 2. Use Type Hints

```python
# Good: Clear types
def parse_user(data: str) -> User:
    return User.from_json(data)

# Bad: No type hints
def parse_user(data):
    return User.from_json(data)
```

### 3. Handle None Values

```python
def safe_transform(item: Optional[str]) -> Optional[str]:
    if item is None:
        return None
    return item.upper()
```

### 4. Batch Operations

```python
class BatchProcessor:
    def __init__(self, batch_size: int = 100):
        self.batch = []
        self.batch_size = batch_size
    
    async def process(self, item: dict) -> Optional[List[dict]]:
        self.batch.append(item)
        
        if len(self.batch) >= self.batch_size:
            result = await self.process_batch(self.batch)
            self.batch = []
            return result
        
        return None  # No output yet
    
    async def flush(self) -> List[dict]:
        if self.batch:
            result = await self.process_batch(self.batch)
            self.batch = []
            return result
        return []
```

## Next Steps

- Learn about [Async Programming](async-programming.md) patterns
- Understand [Backpressure](backpressure.md) handling
- Explore [Lifecycle Management](lifecycle.md)
- See [Common Patterns](patterns.md)