SupplyAgent/api/services/oss_logs.py

"""Query oss_logs for agent run visualizations."""
from __future__ import annotations

from typing import Any

from supply_infra.db.repositories.oss_log_repo import OssLogRepository
from supply_infra.db.session import get_session

_DEMAND_AGENT = "demand_belong_category_agent"


def list_demand_belong_oss_logs() -> list[dict[str, Any]]:
    """List demand_belong_category_agent oss logs, newest first."""
    with get_session() as session:
        repo = OssLogRepository(session)
        rows = repo.list_by_agent_name(_DEMAND_AGENT)
        return [
            {
                "id": row.id,
                "log_name": row.log_name,
                "agent_name": row.agent_name,
                "oss_path": row.oss_path,
                "create_time": row.create_time.isoformat(sep=" ", timespec="seconds")
                if row.create_time
                else None,
            }
            for row in rows
        ]