def retrieve_metadata():
    """Retrieve data from taskinstance, dagrun and serialized dag tables to do some processing to create base tables.

    Returns:
        dataframe: Base tables sla_run_detail and serialized_dags_slas for further processing.
    """
    try:
        pd.set_option("display.max_colwidth", None)

        session = settings.Session()
        taskinstance = session.query(
            TaskInstance.task_id,
            TaskInstance.dag_id,
            TaskInstance.run_id,
            TaskInstance.state,
            TaskInstance.start_date,
            TaskInstance.end_date,
            TaskInstance.duration,
            TaskInstance.operator,
            TaskInstance.queued_dttm,
        ).all()
        taskinstance_df = pd.DataFrame(taskinstance)
        taskinstance_df["run_date"] = pd.to_datetime(taskinstance_df["start_date"]).dt.date
        taskinstance_df["run_date_hour"] = pd.to_datetime(taskinstance_df["start_date"]).dt.hour
        taskinstance_df["task_queue_time"] = (taskinstance_df["start_date"] -
                                              taskinstance_df["queued_dttm"]).dt.total_seconds()
        taskinstance_df = taskinstance_df[taskinstance_df["task_queue_time"] > 0]

        dagrun = session.query(DagRun.dag_id, DagRun.run_id, DagRun.data_interval_end).all()
        dagrun_df = pd.DataFrame(dagrun)
        dagrun_df = dagrun_df.rename(columns={"data_interval_end": "actual_start_time"})

        if "_data" in dir(SerializedDagModel):
            serializeddag = session.query(SerializedDagModel._data).all()
            data_col = "_data"
        else:
            serializeddag = session.query(SerializedDagModel.data).all()
            data_col = "data"

        serializeddag_df = pd.DataFrame(serializeddag)
        serializeddag_json_normalize = pd.json_normalize(
            pd.DataFrame(serializeddag_df[data_col].apply(json.dumps).apply(json.loads).values.tolist())["dag"],
            "tasks", ["_dag_id"])
        serializeddag_filtered = serializeddag_json_normalize[["_dag_id", "task_id", "sla"]]
        serializeddag_filtered = serializeddag_filtered.rename(columns={"_dag_id": "dag_id"})
        serialized_dags_slas = serializeddag_filtered[serializeddag_filtered["sla"].notnull()]

        run_detail = pd.merge(
            dagrun_df[["dag_id", "run_id", "actual_start_time"]],
            taskinstance_df[[
                "task_id",
                "dag_id",
                "run_id",
                "start_date",
                "end_date",
                "duration",
                "task_queue_time",
                "state",
            ]],
            on=["run_id", "dag_id"],
        )

        sla_run = pd.merge(run_detail, serialized_dags_slas, on=["task_id", "dag_id"])
        sla_run_detail = sla_run.loc[sla_run["sla"].isnull() == False]
        sla_run_detail["sla_missed"] = np.where(sla_run_detail["duration"] > sla_run_detail["sla"], 1, 0)
        sla_run_detail["run_date_hour"] = pd.to_datetime(sla_run_detail["start_date"]).dt.hour
        #       sla_run_detail["start_dt"] = sla_run_detail["start_date"].dt.date
        sla_run_detail["start_dt"] = sla_run_detail["start_date"].dt.strftime("%A, %b %d")
        sla_run_detail["start_date"] = pd.to_datetime(sla_run_detail["start_date"]).dt.tz_localize(None)

        return sla_run_detail, serialized_dags_slas

    except:
        no_metadata_found()
Retrieve data from taskinstance, dagrun and serialized dag tables to do some processing to create base tables.

    Returns:
        dataframe: Base tables sla_run_detail and serialized_dags_slas for further processing.
    
sla-miss-report/airflow-sla-miss-report.py