SELECT * FROM events 
WHERE event_date = '{{ ds }}'
AND hour = {{ params.hour }}

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

@dag(
    schedule_interval="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['variable-templates'],
)
def variable_template_dag():
    
    @task
    def use_variable_template():
        """Access variables using Jinja templates."""
        # In operator arguments, use Jinja syntax
        # The actual variable access happens at runtime
        from airflow.models import Variable
        
        # Direct access
        bucket = Variable.get("data_bucket")
        prefix = Variable.get("data_prefix", default_var="raw/")
        
        return f"s3://{bucket}/{prefix}"
    
    @task
    def configure_from_variables():
        """Load configuration from multiple variables."""
        from airflow.models import Variable
        import json
        
        # Load structured config
        config = Variable.get("pipeline_config", deserialize_json=True)
        
        # Access individual settings
        max_retries = Variable.get("max_retries", default_var="3")
        timeout = Variable.get("timeout_seconds", default_var="300")
        
        return {
            "bucket": config.get("bucket"),
            "region": config.get("region"),
            "max_retries": int(max_retries),
            "timeout": int(timeout),
        }
    
    config = configure_from_variables()
    use_variable_template()

variable_template_dag()

from airflow.decorators import task, dag
from datetime import datetime, timedelta
from typing import Dict, Any

@dag(
    schedule_interval="0 6 * * *",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['templating', 'advanced'],
    params={
        "environment": "production",
        "dry_run": False,
        "batch_size": 1000,
    },
)
def advanced_templating_dag():
    
    @task
    def process_with_templates(
        ds: str,
        ds_nodash: str,
        execution_date: datetime,
        prev_execution_date: datetime,
        params: Dict[str, Any],
        **context,
    ):
        """Demonstrate advanced template usage."""
        from airflow.models import Variable
        
        # Access DAG run parameters
        environment = params.get("environment", "development")
        dry_run = params.get("dry_run", False)
        batch_size = params.get("batch_size", 1000)
        
        # Access variables
        config = Variable.get(f"{environment}_config", deserialize_json=True)
        
        # Build dynamic query
        query = f"""
            SELECT * FROM events 
            WHERE event_date >= '{prev_execution_date.strftime("%Y-%m-%d")}'
            AND event_date < '{ds}'
            AND environment = '{environment}'
            LIMIT {batch_size}
        """
        
        print(f"Environment: {environment}")
        print(f"Dry run: {dry_run}")
        print(f"Query:\n{query}")
        
        return {
            "environment": environment,
            "dry_run": dry_run,
            "batch_size": batch_size,
            "query": query,
        }
    
    @task
    def generate_report(config: dict):
        """Generate report with templated content."""
        from airflow.models import Variable
        
        report_template = Variable.get("report_template")
        
        # Use Jinja2 directly for complex templating
        from jinja2 import Template
        
        template = Template(report_template)
        report = template.render(
            environment=config["environment"],
            date=config.get("execution_date"),
            batch_size=config["batch_size"],
        )
        
        return report
    
    config = process_with_templates()
    generate_report(config)

advanced_templating_dag()

# variable_caching.py
from airflow.models import Variable
from airflow import settings
from datetime import datetime, timedelta
import json

class VariableCache:
    """Custom variable caching with TTL."""
    
    def __init__(self, ttl_seconds=300):
        self.ttl = timedelta(seconds=ttl_seconds)
        self.cache = {}
        self.last_access = {}
    
    def get(self, key, default=None, deserialize_json=False):
        """Get variable with caching."""
        now = datetime.now()
        
        # Check cache
        if key in self.cache:
            if now - self.last_access[key] < self.ttl:
                value = self.cache[key]
                if deserialize_json and isinstance(value, str):
                    return json.loads(value)
                return value
        
        # Cache miss - fetch from database
        try:
            value = Variable.get(key, default_var=default)
            self.cache[key] = value
            self.last_access[key] = now
            
            if deserialize_json and isinstance(value, str):
                return json.loads(value)
            return value
        except Exception as e:
            print(f"Error fetching variable {key}: {e}")
            return default
    
    def set(self, key, value, serialize_json=False):
        """Set variable and update cache."""
        if serialize_json:
            value = json.dumps(value)
        
        Variable.set(key, value)
        self.cache[key] = value
        self.last_access[key] = datetime.now()
    
    def invalidate(self, key):
        """Invalidate cached variable."""
        self.cache.pop(key, None)
        self.last_access.pop(key, None)
    
    def clear(self):
        """Clear entire cache."""
        self.cache.clear()
        self.last_access.clear()

# Usage in DAG
variable_cache = VariableCache(ttl_seconds=600)

@dag(schedule_interval="@daily", start_date=datetime(2024, 1, 1))
def cached_variable_dag():
    
    @task
    def process_data():
        """Use cached variable access."""
        config = variable_cache.get("pipeline_config", deserialize_json=True)
        api_key = variable_cache.get("api_key")
        
        print(f"Config: {config}")
        print(f"Using API key: {api_key[:4]}...")
        
        return {"status": "success"}
    
    process_data()

cached_variable_dag()

from airflow.decorators import task, dag
from datetime import datetime
from airflow.models import Variable
import json

# Load environment-specific configuration
ENVIRONMENT = Variable.get("environment", default_var="development")
CONFIG = Variable.get(f"{ENVIRONMENT}_config", deserialize_json=True)

@dag(
    schedule_interval=CONFIG.get("schedule", "@daily"),
    start_date=datetime(2024, 1, 1),
    catchup=CONFIG.get("catchup", False),
    tags=[ENVIRONMENT, 'config-driven'],
    max_active_runs=CONFIG.get("max_active_runs", 1),
)
def config_driven_dag():
    
    @task
    def extract():
        """Extract using environment config."""
        source = CONFIG.get("source")
        connection_id = CONFIG.get(f"{source}_conn_id")
        
        print(f"Extracting from {source} using {connection_id}")
        return {"source": source, "connection_id": connection_id}
    
    @task
    def transform(extract_result: dict):
        """Transform using environment config."""
        transformations = CONFIG.get("transformations", [])
        
        print(f"Applying {len(transformations)} transformations")
        for t in transformations:
            print(f"  - {t}")
        
        return extract_result
    
    @task
    def load(transform_result: dict):
        """Load using environment config."""
        destination = CONFIG.get("destination")
        table = CONFIG.get("load_table")
        
        print(f"Loading to {destination}.{table}")
        return {"status": "success", "records": 0}
    
    extract() >> transform() >> load()

config_driven_dag()