<MathKeyFormula
  title="Dynamic Task Count"
  tex={`N_{\\text{tasks}} = |f_{\\text{map}}(D)|`}
/>

<MathFormula
  title="TaskFlow XCom Throughput"
  tex={`T_{\\text{xcom}} = \\sum_{i=1}^{n} (S_{\\text{push},i} + S_{\\text{pull},i}) \\cdot L_{\\text{latency}}`}
/>

<MathNote type="info">
TaskFlow automatically pushes return values to XCom with key `return_value`. To push multiple values, return a tuple. Each element is stored as a separate XCom with keys `return_value`, `return_value__1`, `return_value__2`, etc.
</MathNote>

<MathNote type="tip">
For large data (>48KB), use `xcom_push` and `xcom_pull` with custom keys, or configure an alternative XCom backend like S3, GCS, or a custom backend. The default database backend has performance limitations for large payloads.
</MathNote>

## Key Concepts Table

| Feature | Traditional Operators | TaskFlow API |
|---------|----------------------|--------------|
| **Task Definition** | `PythonOperator(python_callable=func)` | `@task def func():` |
| **XCom Push** | `ti.xcom_push(key, value)` | `return value` (automatic) |
| **XCom Pull** | `ti.xcom_pull(task_ids, key)` | Function parameter injection |
| **Multiple Returns** | Multiple push calls | Return tuple |
| **DAG Definition** | `with DAG(...):` block | `@dag` decorator |
| **Dynamic Mapping** | `expand()` method | `.map()` on TaskFlow tasks |
| **Code Boilerplate** | High | Low |
| **Type Hints** | Optional | Encouraged for clarity |

## Code Examples

### Advanced TaskFlow Patterns

```python
from airflow.decorators import task, dag
from datetime import datetime, timedelta
from typing import Optional, List, Dict
import json

@dag(
    schedule_interval="0 6 * * *",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['taskflow', 'advanced'],
    doc_md="""
    ## Advanced TaskFlow Patterns
    Demonstrates TaskFlow API with:
    - Multiple return values
    - Dynamic task mapping
    - Error handling
    - Custom XCom keys
    """,
)
def advanced_taskflow_dag():
    
    @task(retries=3, retry_delay=timedelta(minutes=1))
    def extract_from_source(source: str) -> List[Dict]:
        """Extract data with retry logic."""
        import random
        if random.random() < 0.1:
            raise ConnectionError(f"Failed to connect to {source}")
        
        return [
            {"id": i, "source": source, "value": random.randint(1, 100)}
            for i in range(10)
        ]
    
    @task
    def validate_data(records: List[Dict]) -> tuple:
        """Validate and separate valid/invalid records."""
        valid = [r for r in records if 0 <= r["value"] <= 100]
        invalid = [r for r in records if r["value"] < 0 or r["value"] > 100]
        return valid, invalid
    
    @task
    def transform_record(record: Dict) -> Dict:
        """Transform a single record (used with .map())."""
        return {
            **record,
            "value_normalized": record["value"] / 100.0,
            "transformed": True,
        }
    
    @task
    def load_batch(records: List[Dict], destination: str) -> int:
        """Load a batch of records."""
        print(f"Loading {len(records)} records to {destination}")
        return len(records)
    
    @task
    def generate_report(loaded_counts: list, invalid_records: list) -> str:
        """Generate summary report."""
        total_loaded = sum(loaded_counts)
        total_invalid = len(invalid_records)
        
        report = {
            "total_loaded": total_loaded,
            "total_invalid": total_invalid,
            "success_rate": total_loaded / (total_loaded + total_invalid) * 100,
        }
        return json.dumps(report, indent=2)
    
    # Define sources
    sources = ["postgres", "mysql", "mongodb"]
    
    # Extract from multiple sources (dynamic mapping)
    raw_data = extract_from_source.map(sources)
    
    # Validate each source's data
    valid_data, invalid_data = validate_data.expand(raw_data)
    
    # Transform valid records (dynamic mapping)
    transformed = transform_record.map(valid_data)
    
    # Load to destination
    loaded_count = load_batch(transformed, "data_warehouse")
    
    # Generate report
    generate_report(loaded_count, invalid_data)

advanced_taskflow_dag()