Orchestration: Apache Airflow, CI/CD & Governance

from datetime import datetime, timedelta
from airflow import DAG
from airflow.sensors.filesystem import FileSensor
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.operators.python import PythonOperator
from airflow.utils.task_group import TaskGroup

def alert_slack_failure(context):
    """Callback posting task failure alert with execution metadata to Slack."""
    task_id = context['task_instance'].task_id
    exec_date = context['execution_date']
    print(f"CRITICAL: Task {task_id} failed for execution date {exec_date}. Alert sent.")

default_args = {
    'owner': 'pranay_sarode',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['padmabushanenterpirses@pranaysarode.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=3),
    'retry_exponential_backoff': True,
    'on_failure_callback': alert_slack_failure,
}

with DAG(
    dag_id='daily_capital_group_reconciliation',
    default_args=default_args,
    description='Reconcile internal trades vs custodian positions with automated DLQ',
    schedule_interval='0 6 * * 1-5', # 06:00 UTC Monday-Friday
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['production', 'financial', 'capital_group']
) as dag:

    # 1. Wait for Custodian Settlement file on S3 in Reschedule mode
    wait_for_custodian_file = S3KeySensor(
        task_id='wait_for_custodian_settlement',
        bucket_name='enterprise-custodian-landing',
        bucket_key='settlement_.csv',
        mode='reschedule',          # Releases worker slot while sleeping
        poke_interval=120,          # Check every 2 minutes
        timeout=3600                # Timeout after 1 hour
    )

    with TaskGroup("reconciliation_group") as recon_group:
        def run_pyspark_reconciliation():
            print("Executing PySpark 3-Way NAV Reconciliation Job...")

        execute_recon = PythonOperator(
            task_id='execute_3way_recon',
            python_callable=run_pyspark_reconciliation
        )

    wait_for_custodian_file >> recon_group