SupplyAgent/api/services/scheduler.py

"""Scheduler API helpers."""
from __future__ import annotations

from typing import Any

from supply_infra.scheduler.constants import SUPPLY_PIPELINE_JOB_ID
from supply_infra.scheduler.manual_jobs import (
    get_manual_run,
    list_manual_jobs,
    trigger_manual_job,
)


def list_triggerable_jobs() -> list[dict[str, Any]]:
    return list_manual_jobs()


def run_scheduler_job(
    job_id: str,
    *,
    biz_dt: str | None = None,
    partition_date: str | None = None,
    wait: bool = False,
) -> dict[str, Any]:
    return trigger_manual_job(
        job_id,
        biz_dt=biz_dt,
        partition_date=partition_date,
        wait=wait,
    )


def get_scheduler_job_run(run_id: str) -> dict[str, Any] | None:
    return get_manual_run(run_id)


def run_supply_pipeline(
    *,
    biz_dt: str | None = None,
) -> dict[str, Any]:
    """异步执行供给数据全流程，立即返回 run_id。"""
    return run_scheduler_job(
        SUPPLY_PIPELINE_JOB_ID,
        biz_dt=biz_dt,
        wait=False,
    )