Add job history streaming endpoint

autosubmit_api/repositories/job_data.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import Any, List, Union
+from typing import Any, List, Union, Generator
 from pydantic import BaseModel
 from sqlalchemy import Engine, Table, inspect, or_, Index
 from sqlalchemy.schema import CreateTable
@@ -84,6 +84,12 @@ def get_job_data_COMPLETD_by_section(
         Gets job data by section
         """
 
+    @abstractmethod
+    def get_all_generator(self) -> Generator[ExperimentJobDataModel, Any, None]:
+        """
+        Gets all job data as generator
+        """
+
 
 class ExperimentJobDataSQLRepository(ExperimentJobDataRepository):
     def __init__(self, expid: str, engine: Engine, valid_tables: List[Table]):
@@ -217,6 +223,16 @@ def get_job_data_COMPLETD_by_section(self, section: str):
             for row in result
         ]
 
+    def get_all_generator(self):
+        with self.engine.connect() as conn:
+            statement = (
+                self.table.select().where(self.table.c.id > 0).order_by(self.table.c.id)
+            )
+            result = conn.execute(statement)
+
+            for row in result:
+                yield ExperimentJobDataModel.model_validate(row, from_attributes=True)
+
 
 def create_experiment_job_data_repository(expid: str):
     engine = create_sqlite_db_engine(ExperimentPaths(expid).job_data_db)

autosubmit_api/routers/v4/experiments.py

@@ -1,20 +1,23 @@
 import asyncio
 from collections import deque
+import csv
 from datetime import datetime, timezone
 from http import HTTPStatus
+from io import StringIO
 import json
 import math
 import traceback
 from typing import Annotated, Any, Dict, List, Literal, Optional
 from fastapi import APIRouter, Depends, HTTPException, Query
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse, StreamingResponse
 from autosubmit_api.auth import auth_token_dependency
 from autosubmit_api.builders.experiment_builder import ExperimentBuilder
 from autosubmit_api.config.config_common import AutosubmitConfigResolver
 from autosubmit_api.database import tables
 from autosubmit_api.common.utils import Status
 from autosubmit_api.database.db_jobdata import JobDataStructure
 from autosubmit_api.database.models import BaseExperimentModel
+from autosubmit_api.repositories.job_data import create_experiment_job_data_repository
 from autosubmit_api.repositories.join.experiment_join import (
     create_experiment_join_repository,
 )
@@ -344,3 +347,69 @@ async def get_run_config(
         "config": _format_config_response(metadata),
     }
     return response
+
+
+@router.get("/{expid}/jobs-history", name="Get experiment jobs history")
+async def get_jobs_history(
+    expid: str,
+    format: Annotated[Literal["json", "csv"], Query()] = "json",
+    user_id: Optional[str] = Depends(auth_token_dependency()),
+) -> Dict[str, Any]:
+    """
+    Get the jobs history of an experiment
+    """
+
+    def _generate_csv(expid: str):
+        job_generator = create_experiment_job_data_repository(expid).get_all_generator()
+
+        csv_file = StringIO()
+        writer = csv.writer(csv_file)
+
+        # Yield the header first
+        header = [
+            "job_name",
+            "run_id",
+            "section",
+            "status",
+            "submit",
+            "start",
+            "finish",
+            "out",
+            "err",
+        ]
+        writer.writerow(header)
+        csv_file.seek(0)
+        yield csv_file.read()
+        csv_file.seek(0)
+        csv_file.truncate(0)
+
+        # Yield each job row
+        for job in job_generator:
+            writer.writerow(
+                [
+                    job.job_name,
+                    job.run_id,
+                    job.section,
+                    job.status,
+                    job.submit,
+                    job.start,
+                    job.finish,
+                    job.out,
+                    job.err,
+                ]
+            )
+            csv_file.seek(0)
+            yield csv_file.read()
+            csv_file.seek(0)
+            csv_file.truncate(0)
+
+    if format == "csv":
+        response = StreamingResponse(_generate_csv(expid), media_type="text/csv")
+        response.headers["Content-Disposition"] = (
+            f"attachment; filename=jobs_history_{expid}.csv"
+        )
+        return response
+
+    raise HTTPException(
+        status_code=HTTPStatus.NOT_IMPLEMENTED, detail="Only format=csv is supported"
+    )

tests/test_endpoints_v4.py

@@ -351,3 +351,25 @@ def test_run_config_v3_retro(self, run_id: int, fixture_fastapi_client: TestClie
         for key in ALLOWED_CONFIG_KEYS:
             assert key in resp_obj["config"]
             assert isinstance(resp_obj["config"][key], dict)
+
+
+class TestJobHistory:
+    endpoint = "/v4/experiments/{expid}/jobs-history"
+
+    def test_csv_stream_response(self, fixture_fastapi_client: TestClient):
+        expid = "a3tb"
+        response = fixture_fastapi_client.get(
+            self.endpoint.format(expid=expid),
+            params={"format": "csv"},
+        )
+
+        assert response.status_code == HTTPStatus.OK
+        assert response.headers["Content-Type"] == "text/csv; charset=utf-8"
+        assert response.headers["Content-Disposition"].startswith(
+            "attachment; filename="
+        )
+        assert response.headers["Content-Disposition"].endswith(".csv")
+
+        csv_content = response.content.decode("utf-8")
+        assert len(csv_content) > 0
+        assert csv_content.count("\n") > 1