from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
import pandas as pd

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email': ['alerts@company.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'retry_exponential_backoff': True,
}

dag = DAG(
    'daily_sales_pipeline',
    default_args=default_args,
    description='Process daily sales data',
    schedule_interval='0 2 * * *',  # Run at 2 AM daily
    catchup=False,
    tags=['sales', 'etl'],
)

def extract_data(**context):
    """Extract data from source"""
    # Idempotent: check if already processed
    execution_date = context['ds']
    print(f"Extracting data for {execution_date}")
    # Your extraction logic here
    return f"data_{execution_date}.csv"

def transform_data(**context):
    """Transform the extracted data"""
    ti = context['ti']
    filename = ti.xcom_pull(task_ids='extract')

    df = pd.read_csv(filename)
    # Transformation logic
    df['total'] = df['quantity'] * df['price']

    output_file = f"transformed_{context['ds']}.csv"
    df.to_csv(output_file, index=False)
    return output_file

def load_data(**context):
    """Load data to warehouse"""
    ti = context['ti']
    filename = ti.xcom_pull(task_ids='transform')
    print(f"Loading {filename} to warehouse")
    # Load logic here

undefined

undefined

undefined

#!/usr/bin/env python3
"""
Scheduled backup script with retry logic
Add to crontab: 0 3 * * * /usr/bin/python3 /path/to/backup.py
"""

import subprocess
import time
import logging
from datetime import datetime
import smtplib
from email.message import EmailMessage

logging.basicConfig(
    filename='/var/log/backup.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def send_alert(subject, message):
    """Send email alert on failure"""
    msg = EmailMessage()
    msg.set_content(message)
    msg['Subject'] = subject
    msg['From'] = 'backup@company.com'
    msg['To'] = 'admin@company.com'

    try:
        with smtplib.SMTP('localhost') as s:
            s.send_message(msg)
    except Exception as e:
        logging.error(f"Failed to send alert: {e}")

def backup_database(max_retries=3):
    """Backup database with retry logic"""
    backup_file = f"/backups/db_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.sql"

    for attempt in range(1, max_retries + 1):
        try:
            logging.info(f"Starting backup attempt {attempt}/{max_retries}")

            # Run backup command
            result = subprocess.run(
                ['pg_dump', '-U', 'postgres', '-d', 'mydb', '-f', backup_file],
                capture_output=True,
                text=True,
                timeout=300
            )

            if result.returncode == 0:
                logging.info(f"Backup successful: {backup_file}")

                # Verify backup file exists and has content
                import os
                if os.path.exists(backup_file) and os.path.getsize(backup_file) > 0:
                    return True
                else:
                    raise Exception("Backup file is empty or missing")
            else:
                raise Exception(f"Backup command failed: {result.stderr}")

        except Exception as e:
            logging.error(f"Backup attempt {attempt} failed: {e}")

            if attempt < max_retries:
                # Exponential backoff
                wait_time = 2 ** attempt
                logging.info(f"Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                # Final failure
                error_msg = f"Backup failed after {max_retries} attempts"
                logging.critical(error_msg)
                send_alert("CRITICAL: Database Backup Failed", error_msg)
                return False

    return False

if __name__ == '__main__':
    logging.info("=== Backup job started ===")
    success = backup_database()

    if success:
        logging.info("=== Backup job completed successfully ===")
        exit(0)
    else:
        logging.error("=== Backup job failed ===")
        exit(1)