from typing import Any, Callable, Dict

from airflow import DAG
from twitch_airflow_components.dag_utils import pagerduty_on_failure_callback

from conductor.config import Project
from conductor.operator_registry import OperatorRegistry

PAGERDUTY_PARAM = "pagerduty_conn_id"


class Conductor:
    def __init__(
        self,
        name: str,
        project: Project,
        on_failure_callback: Callable = None,
        params: Dict[str, Any] = None,
        **kwargs
    ) -> None:
        self._name = name

        # Initialize DAG and DAG Resources.
        self._dag_resources = project.get_dag_resources(name)

        if params is None:
            params = {}

        # Create a default pagerduty failure handler
        if on_failure_callback is None and project.project_resources.env.prod:
            if PAGERDUTY_PARAM not in params:
                pagerduty_config = project.project_resources.env.pagerduty
                pagerduty_conn_id = (
                    "pagerduty"
                    if pagerduty_config is None
                    else pagerduty_config.conn_id
                )
                params[PAGERDUTY_PARAM] = pagerduty_conn_id
            on_failure_callback = pagerduty_on_failure_callback

        self._dag = DAG(
            self._dag_resources.dag_id(),
            on_failure_callback=on_failure_callback,
            params=params,
            **kwargs
        )
        self.task_container_params: Dict[str, Any] = {}
        self.operators = OperatorRegistry(
            project.project_resources,
            self._dag_resources,
            self.task_container_params,
            self._dag,
        )

    @property
    def name(self) -> str:
        return self._name

    @property
    def dag(self) -> DAG:
        return self._dag

    @property
    def dag_resources(self) -> DAG:
        return self._dag_resources
