@dag(schedule_interval=timedelta(hours=1), start_date=datetime(2024, 1, 1), catchup=False)
def taskflow_example():
    @task
    def extract_data():
        return [{"id": 1, "name": "Alice"}]  # Auto-pushed to XCom

    @task
    def transform_data(raw_data: list) -> list:  # Auto-pulled from XCom
        return [r.upper() for r in raw_data]

    raw = extract_data()
    transformed = transform_data(raw)  # Dependency auto-inferred

@task
def extract_metrics() -> Tuple[List[Dict], Dict]:
    return records, metadata  # Each pushed as separate XCom key

@task
def process_records(records: list, metadata: dict):  # Pull by index
    print(f"Processing {metadata['total_records']} records")

@task
def get_regions() -> list:
    return ["us-east-1", "us-west-2", "eu-west-1"]

@task
def process_region(region: str) -> dict:
    return {"region": region, "status": "success"}

@task
def aggregate_results(region_results: list) -> dict:
    return {"total_regions": len(region_results)}

# Usage
regions = get_regions()
results = process_region.map(regions)  # Dynamic mapping
aggregate_results(results)

from airflow.decorators import task, dag
from airflow.models.xcom import BaseXCom
from datetime import datetime, timedelta
import json
import pickle

class CustomXComBackend(BaseXCom):
    """Custom XCom backend for large payloads."""
    
    @staticmethod
    def serialize_value(value):
        """Serialize value for storage."""
        if isinstance(value, (dict, list)):
            return json.dumps(value)
        return str(value)
    
    @staticmethod
    def deserialize_value(result):
        """Deserialize value from storage."""
        try:
            return json.loads(result)
        except (json.JSONDecodeError, TypeError):
            return result

@dag(
    schedule_interval="@hourly",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['taskflow', 'xcom-backend'],
    render_template_as_native_obj=True,
)
def xcom_backend_dag():
    
    @task(
        xcom_backend='path.to.CustomXComBackend',
        pool='data_processing',
    )
    def process_large_dataset(records: list) -> dict:
        """Process dataset using custom XCom backend."""
        import pandas as pd
        
        df = pd.DataFrame(records)
        
        result = {
            "mean": float(df["value"].mean()),
            "std": float(df["value"].std()),
            "count": int(len(df)),
            "percentiles": {
                "25": float(df["value"].quantile(0.25)),
                "50": float(df["value"].quantile(0.50)),
                "75": float(df["value"].quantile(0.75)),
            },
        }
        return result
    
    @task
    def summarize(stats: dict) -> str:
        """Create summary string."""
        return f"Processed {stats['count']} records, mean={stats['mean']:.2f}"
    
    sample_data = [{"value": i * 1.5} for i in range(100)]
    stats = process_large_dataset(sample_data)
    summarize(stats)

xcom_backend_dag()

from airflow.decorators import task, dag
from datetime import datetime, timedelta

@dag(
    schedule_interval="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['taskflow', 'taskmap', 'cross-dependencies'],
)
def cross_task_dependency_dag():
    
    @task
    def get_config() -> dict:
        """Return configuration for downstream tasks."""
        return {
            "environments": ["dev", "staging", "prod"],
            "parallel_workers": 4,
            "timeout_seconds": 300,
        }
    
    @task
    def setup_environment(env: str, config: dict) -> dict:
        """Setup each environment using config."""
        return {
            "environment": env,
            "status": "ready",
            "workers": config["parallel_workers"],
        }
    
    @task
    def run_tests(env_setup: dict) -> dict:
        """Run tests in each environment."""
        return {
            "environment": env_setup["environment"],
            "tests_passed": 42,
            "tests_failed": 0,
        }
    
    @task
    def cleanup_environments(results: list) -> str:
        """Cleanup all environments after tests."""
        environments = [r["environment"] for r in results]
        return f"Cleaned up: {', '.join(environments)}"
    
    # Get configuration
    config = get_config()
    
    # Setup environments dynamically using config
    env_setups = setup_environment.map(
        env=config["environments"],
        config=config,  # Same config passed to all mapped tasks
    )
    
    # Run tests in each environment
    test_results = run_tests.expand(env_setups)
    
    # Cleanup
    cleanup_environments(test_results)

cross_task_dependency_dag()