# Shell Commands

Flokkit Flow provides seamless integration with shell commands, allowing you to use external programs as nodes in your flow graphs. This enables powerful data processing pipelines that leverage existing command-line tools.

## Overview

Shell commands can be integrated as:
- **Sources**: Stream output from commands
- **Transforms**: Process data through commands
- **Sinks**: Send data to commands

All shell integrations are:
- Type-safe
- Async-native
- Support streaming
- Handle backpressure
- Provide proper error handling

## Basic Usage

### Shell Command as Source

Use shell commands to generate or read data:

```python
from flow import flow
from flow.shell import shell_source

# Stream lines from a file
cat_source = shell_source("cat /var/log/app.log")

await (
    flow()
    .source(cat_source, str)
    .transform(parse_log_line, LogEntry)
    .sink(store_in_db)
    .execute()
)

# Stream real-time logs
tail_logs = shell_source("tail -f /var/log/syslog")

await (
    flow()
    .source(tail_logs, str)
    .filter(lambda line: "ERROR" in line)
    .sink(alert_on_error)
    .execute()
)
```

### Shell Command as Transform

Process data through external commands:

```python
from flow.shell import shell_transform

# Text transformation with sed
clean_text = shell_transform("sed 's/[^a-zA-Z0-9 ]//g'")

await (
    flow()
    .source(dirty_texts, str)
    .transform(clean_text, str)
    .sink(save_clean_text)
    .execute()
)

# JSON processing with jq
extract_field = shell_transform("jq -r '.user.name'")

await (
    flow()
    .source(json_documents, str)
    .transform(extract_field, str)
    .sink(collect_names)
    .execute()
)
```

### Shell Command as Sink

Send data to external commands:

```python
from flow.shell import shell_sink

# Append to log file
append_log = shell_sink("tee -a application.log")

await (
    flow()
    .source(events, str)
    .transform(format_event, str)
    .sink(append_log)
    .execute()
)

# Send notifications
notify = shell_sink("notify-send 'Alert'", input_mode="args")

await (
    flow()
    .source(alerts, str)
    .sink(notify)
    .execute()
)
```

## Advanced Features

### The ShellCommand Class

For more control, use the `ShellCommand` class directly:

```python
from flow.shell import ShellCommand

cmd = ShellCommand(
    "grep -E 'pattern'",
    input_mode="stdin",      # How to pass input: "stdin", "args", or "none"
    output_mode="lines",     # How to capture output: "lines", "all", or "none"
    cwd="/path/to/dir",      # Working directory
    env={"VAR": "value"},    # Environment variables
    timeout=30.0,            # Command timeout in seconds
    check=True,              # Raise on non-zero exit
)

# Use as transform
result = await cmd("input text")
```

### Input Modes

Control how data is passed to commands:

```python
# stdin mode - data passed via stdin
grep = shell_transform("grep ERROR", input_mode="stdin")

# args mode - data passed as command arguments
echo = shell_transform("echo", input_mode="args")

# none mode - no input (useful for sources)
date = shell_source("date +%s", input_mode="none")
```

### Output Modes

Control how output is captured:

```python
# lines mode - output split by lines (default)
lines = shell_source("ls -1", output_mode="lines")

# all mode - entire output as single string
content = shell_transform("cat", output_mode="all")

# none mode - no output captured (useful for sinks)
writer = shell_sink("tee output.txt", output_mode="none")
```

### Custom Parsers

Parse command output into custom types:

```python
import json

# Parse JSON output
def parse_json(text: str) -> dict:
    return json.loads(text)

json_source = shell_source(
    "curl -s https://api.example.com/data",
    dict,
    parser=parse_json
)

# Parse numbers
def parse_int(text: str) -> int:
    return int(text.strip())

number_gen = shell_source(
    "seq 1 10",
    int,
    parser=parse_int
)
```

### Error Handling

Handle command failures gracefully:

```python
# Don't raise on failure
cmd = ShellCommand("grep pattern", check=False)

async def safe_grep(text: str) -> str:
    result = await cmd(text)
    return result or "No match found"

# With timeout
risky_cmd = ShellCommand(
    "slow-command",
    timeout=5.0  # Kill after 5 seconds
)

try:
    result = await risky_cmd.run()
except TimeoutError:
    print("Command timed out")
```

## Complex Pipelines

Build sophisticated pipelines combining multiple shell commands:

```python
# Data processing pipeline
read_csv = shell_source("csvcut -c name,age,city data.csv")
filter_adults = shell_transform("awk -F, '$2 >= 18'")
sort_by_name = shell_transform("sort -t, -k1")
to_json = shell_transform("csvjson", parser=json.loads)

results = []
await (
    flow("CSV Processing")
    .source(read_csv, str)
    .transform(filter_adults, str)
    .transform(sort_by_name, str)
    .transform(to_json, dict)
    .sink(results.append)
    .execute()
)
```

## Best Practices

### Security

1. **Avoid shell injection**: Never pass untrusted input directly to shell commands
2. **Use argument lists**: Prefer list form over string commands
3. **Validate input**: Always validate data before passing to commands

```python
# BAD - vulnerable to injection
user_input = "file.txt; rm -rf /"
cmd = shell_source(f"cat {user_input}")  # DANGER!

# GOOD - safe from injection
cmd = ShellCommand(["cat", user_input])  # Safe
```

### Performance

1. **Stream large data**: Use streaming for large files
2. **Set timeouts**: Prevent hanging on slow commands
3. **Handle backpressure**: Commands respect flow backpressure

```python
# Stream large files efficiently
large_file = shell_source("zcat huge.gz")

await (
    flow()
    .source(large_file, str)
    .batch(1000)  # Process in batches
    .transform(process_batch, List[str])
    .sink(bulk_insert)
    .execute()
)
```

### Error Handling

1. **Check return codes**: Use `check=True` for critical commands
2. **Handle timeouts**: Set reasonable timeouts
3. **Log stderr**: Capture error output for debugging

```python
cmd = ShellCommand(
    "important-command",
    check=True,
    timeout=30.0
)

try:
    result = await cmd.run()
    if result.stderr:
        logger.warning(f"Command warnings: {result.stderr}")
except RuntimeError as e:
    logger.error(f"Command failed: {e}")
except TimeoutError:
    logger.error("Command timed out")
```

## Examples

### Log Analysis Pipeline

```python
# Analyze web server logs
async def analyze_logs():
    # Source: Read compressed logs
    read_logs = shell_source("zcat /var/log/nginx/access.log.*.gz")
    
    # Transform: Extract timestamps and status codes
    extract_fields = shell_transform(
        "awk '{print $4, $9}'",
        parser=lambda s: s.strip().split()
    )
    
    # Transform: Filter 5xx errors
    filter_errors = shell_transform(
        "grep ' 5[0-9][0-9]$'",
        check=False  # Don't fail if no matches
    )
    
    errors = []
    await (
        flow("Log Analysis")
        .source(read_logs, str)
        .transform(extract_fields, list)
        .filter(lambda x: len(x) == 2)
        .transform(lambda x: f"{x[0]} {x[1]}", str)
        .transform(filter_errors, str)
        .sink(errors.append)
        .execute()
    )
    
    print(f"Found {len(errors)} 5xx errors")
```

### Data ETL Pipeline

```python
# Extract, transform, and load data
async def etl_pipeline():
    # Extract: Download data
    download = shell_source(
        "curl -s https://example.com/data.json | jq -c '.[]'"
    )
    
    # Transform: Process with Python
    async def enrich_record(json_str: str) -> dict:
        record = json.loads(json_str)
        record['processed_at'] = datetime.now().isoformat()
        return record
    
    # Load: Insert into database (via CLI tool)
    insert_cmd = shell_sink(
        "psql -c 'INSERT INTO records VALUES (%s)'",
        input_mode="args"
    )
    
    await (
        flow("ETL Pipeline")
        .source(download, str)
        .transform(enrich_record, dict)
        .transform(json.dumps, str)
        .sink(insert_cmd)
        .execute()
    )
```

## Integration with Middleware

Shell commands work seamlessly with Flokkit Flow's middleware system:

```python
from flow import LoggingMiddleware, RetryMiddleware

# Add logging and retry to shell commands
logger = LoggingMiddleware()
retry = RetryMiddleware(max_attempts=3)

await (
    flow()
    .with_middleware(logger, retry)
    .source(shell_source("unstable-api-call"), str)
    .transform(parse_response, dict)
    .sink(save_result)
    .execute()
)
```

## Limitations

1. **Platform dependent**: Commands must exist on the system
2. **Text-based**: Shell I/O is text-based (binary needs encoding)
3. **Process overhead**: Each command spawns a new process
4. **Security risks**: Be careful with untrusted input

## Summary

Shell command integration in Flokkit Flow provides:
- Seamless integration with existing tools
- Type-safe command execution
- Full async/streaming support
- Flexible input/output modes
- Robust error handling

This enables building powerful data pipelines that leverage the entire ecosystem of command-line tools while maintaining the safety and composability of Flokkit Flow.