"""
Template for an Airflow DAG that runs a dlt connector via DockerOperator.

Copy this file to dags/dag_extract_load__<connector_name>.py and customize.
"""

import pendulum

from airflow.sdk import dag, task
from airflow.providers.docker.operators.docker import DockerOperator

from utils.common import get_dlt_destination, get_dlt_secrets_toml_base64


@dag(
    schedule=None,  # Set schedule: "@daily", "@hourly", or cron expression
    start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
    catchup=False,
    tags=["extract_load", "dlt"],
)
def extract_load_CONNECTOR_NAME():

    @task()
    def run_dlt_pipeline(destination: str, secrets_toml_base64: str):
        return DockerOperator(
            task_id="run_dlt_pipeline",
            image="dlt-connector-CONNECTOR_NAME:latest",
            container_name="dlt-CONNECTOR_NAME",
            api_version="auto",
            auto_remove="force",
            docker_url="tcp://docker-proxy:2375",
            environment={
                "DLT_DESTINATION": destination,
                "DLT_SOURCE_NAME": "CONNECTOR_NAME",
                "DLT_SECRETS_TOML_BASE64": secrets_toml_base64,
            },
        ).execute(context={})

    secrets_toml_base64 = get_dlt_secrets_toml_base64()
    destination = get_dlt_destination()
    run_dlt_pipeline(destination, secrets_toml_base64)


dag = extract_load_CONNECTOR_NAME()