添加查看调度运行结果

backend/src/backend/schedule_runs.py

@@ -1,9 +1,20 @@
 from __future__ import annotations
 
+import asyncio
 from datetime import UTC, datetime
 from typing import Any, Literal
+from urllib.parse import quote
 
-from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, status
+from fastapi import (
+    APIRouter,
+    Depends,
+    Header,
+    HTTPException,
+    Query,
+    Request,
+    Response,
+    status,
+)
 from pydantic import Field
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -11,6 +22,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from common.db.models import (
     ScheduleNodeRuns,
     ScheduleRuns,
+    StorageObjects,
 )
 from backend.dependencies import (
     RequestContext,
@@ -149,6 +161,75 @@ async def _visible_run(
     return item
 
 
+async def _visible_node_run(
+    run_id: str,
+    node_run_id: str,
+    context: RequestContext,
+    session: AsyncSession,
+) -> ScheduleNodeRuns:
+    await _visible_run(run_id, context, session)
+    item = await session.scalar(
+        select(ScheduleNodeRuns).where(
+            ScheduleNodeRuns.node_run_id == node_run_id,
+            ScheduleNodeRuns.run_id == run_id,
+        )
+    )
+    if item is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "schedule node run not found")
+    return item
+
+
+async def _visible_artifact(
+    storage_object_id: str | None,
+    *,
+    usage_type: Literal["run_log", "run_result"],
+    context: RequestContext,
+    session: AsyncSession,
+) -> StorageObjects | None:
+    if storage_object_id is None:
+        return None
+    item = await session.scalar(
+        select(StorageObjects).where(
+            StorageObjects.storage_object_id == storage_object_id,
+            StorageObjects.workspace_id == context.workspace.workspace_id,
+            StorageObjects.usage_type == usage_type,
+            StorageObjects.object_status == "available",
+            StorageObjects.is_deleted == 0,
+        )
+    )
+    if (
+        item is None
+        or item.storage_backend != "rustfs"
+        or not item.bucket_name
+        or not item.object_key
+    ):
+        return None
+
+    return item
+
+
+def _artifact_payload(item: StorageObjects | None, *, url: str) -> dict[str, Any] | None:
+    if item is None:
+        return None
+    return {
+        "url": url,
+        "file_name": item.file_name,
+        "mime_type": item.mime_type,
+        "size_bytes": item.size_bytes,
+    }
+
+
+async def _artifact_bytes(
+    item: StorageObjects,
+    request: Request,
+) -> bytes:
+    return await asyncio.to_thread(
+        request.app.state.object_store.get_bytes,
+        bucket_name=item.bucket_name,
+        object_key=item.object_key,
+    )
+
+
 @router.post(
     "/api/v1/schedules/{schedule_id}/run",
     status_code=status.HTTP_202_ACCEPTED,
@@ -240,3 +321,101 @@ async def get_schedule_run(
         "data": await run_detail(item, session),
         "meta": {},
     }
+
+
+@router.get(
+    "/api/v1/schedule-runs/{run_id}/node-runs/{node_run_id}/artifacts"
+)
+async def get_schedule_node_run_artifacts(
+    run_id: str,
+    node_run_id: str,
+    context: RequestContext = Depends(request_context),
+    session: AsyncSession = Depends(database_session),
+) -> dict[str, Any]:
+    node_run = await _visible_node_run(run_id, node_run_id, context, session)
+    log_artifact = await _visible_artifact(
+        node_run.logs_object_id,
+        usage_type="run_log",
+        context=context,
+        session=session,
+    )
+    result_artifact = await _visible_artifact(
+        node_run.result_object_id,
+        usage_type="run_result",
+        context=context,
+        session=session,
+    )
+    base_path = f"/api/v1/schedule-runs/{run_id}/node-runs/{node_run_id}"
+    workspace_query = f"workspace_id={context.workspace.workspace_id}"
+    return {
+        "request_id": context.request_id,
+        "data": {
+            "run_id": run_id,
+            "node_run_id": node_run_id,
+            "log": _artifact_payload(
+                log_artifact,
+                url=f"{base_path}/logs?{workspace_query}",
+            ),
+            "result": _artifact_payload(
+                result_artifact,
+                url=f"{base_path}/result?{workspace_query}",
+            ),
+        },
+        "meta": {},
+    }
+
+
+@router.get(
+    "/api/v1/schedule-runs/{run_id}/node-runs/{node_run_id}/logs"
+)
+async def read_schedule_node_run_logs(
+    run_id: str,
+    node_run_id: str,
+    request: Request,
+    context: RequestContext = Depends(request_context),
+    session: AsyncSession = Depends(database_session),
+) -> Response:
+    node_run = await _visible_node_run(run_id, node_run_id, context, session)
+    item = await _visible_artifact(
+        node_run.logs_object_id,
+        usage_type="run_log",
+        context=context,
+        session=session,
+    )
+    if item is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "schedule node log not found")
+    return Response(
+        content=await _artifact_bytes(item, request),
+        media_type=item.mime_type or "text/plain; charset=utf-8",
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+@router.get(
+    "/api/v1/schedule-runs/{run_id}/node-runs/{node_run_id}/result"
+)
+async def download_schedule_node_run_result(
+    run_id: str,
+    node_run_id: str,
+    request: Request,
+    context: RequestContext = Depends(request_context),
+    session: AsyncSession = Depends(database_session),
+) -> Response:
+    node_run = await _visible_node_run(run_id, node_run_id, context, session)
+    item = await _visible_artifact(
+        node_run.result_object_id,
+        usage_type="run_result",
+        context=context,
+        session=session,
+    )
+    if item is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "schedule node result not found")
+    encoded_name = quote(item.file_name, safe="")
+    return Response(
+        content=await _artifact_bytes(item, request),
+        media_type=item.mime_type or "application/octet-stream",
+        headers={
+            "Cache-Control": "no-store",
+            "Content-Disposition": f"attachment; filename*=UTF-8''{encoded_name}",
+        },
+    )