from __future__ import annotations

from pathlib import Path

import pendulum
from airflow.sdk import DAG, task

from orchestration.airflow.engine.python_assets import (
    attach_output_row_count_metadata,
    build_airflow_asset,
    load_python_asset_job_config,
    run_python_asset_job as execute_python_asset_job,
)


PYTHON_REVREC_CONFIG_PATH = Path(__file__).resolve().parents[1] / "sources" / "python-revrec.yaml"
PYTHON_REVREC = load_python_asset_job_config(PYTHON_REVREC_CONFIG_PATH)
PYTHON_REVREC_INLETS = [build_airflow_asset(asset) for asset in PYTHON_REVREC.inlets]
PYTHON_REVREC_OUTLETS = [build_airflow_asset(asset) for asset in PYTHON_REVREC.outlets]


with DAG(
    dag_id=PYTHON_REVREC.dag_id,
    default_args={"owner": PYTHON_REVREC.owner},
    schedule=PYTHON_REVREC_INLETS,
    start_date=pendulum.parse(PYTHON_REVREC.start_date),
    catchup=PYTHON_REVREC.catchup,
    max_active_runs=PYTHON_REVREC.max_active_runs,
    tags=list(PYTHON_REVREC.tags),
) as dag:

    @task(
        task_id=PYTHON_REVREC.task_id,
        inlets=PYTHON_REVREC_INLETS,
        outlets=PYTHON_REVREC_OUTLETS,
    )
    def run_python_asset_job(outlet_events=None) -> dict[str, object]:
        metadata = execute_python_asset_job(PYTHON_REVREC)
        attach_output_row_count_metadata(
            outlet_events=outlet_events,
            airflow_assets=PYTHON_REVREC_OUTLETS,
            metadata=metadata,
        )
        return metadata

    run_python_asset_job()