Improve sources and runs history tables

repub/jobs.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import math
 import os
 import signal
 import subprocess
@@ -10,7 +11,7 @@ import time
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
-from typing import Callable, TextIO, cast
+from typing import Callable, TextIO, TypedDict, cast
 
 from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.cron import CronTrigger
@@ -30,6 +31,7 @@ from repub.model import (
 SCHEDULER_JOB_PREFIX = "job-"
 POLL_JOB_ID = "runtime-poll-workers"
 SYNC_JOB_ID = "runtime-sync-jobs"
+COMPLETED_EXECUTION_PAGE_SIZE = 20
 
 
 @dataclass(frozen=True)
@@ -102,6 +104,17 @@ class ExecutionLogView:
     error_message: str | None = None
 
 
+class RunsView(TypedDict):
+    running: tuple[dict[str, object], ...]
+    queued: tuple[dict[str, object], ...]
+    upcoming: tuple[dict[str, object], ...]
+    completed: tuple[dict[str, object], ...]
+    completed_page: int
+    completed_page_size: int
+    completed_total_count: int
+    completed_total_pages: int
+
+
 class JobRuntime:
     def __init__(
         self,
@@ -647,10 +660,15 @@ class JobRuntime:
 
 
 def load_runs_view(
-    *, log_dir: str | Path, now: datetime | None = None
-) -> dict[str, tuple[dict[str, object], ...]]:
+    *,
+    log_dir: str | Path,
+    now: datetime | None = None,
+    completed_page: int = 1,
+    completed_page_size: int = COMPLETED_EXECUTION_PAGE_SIZE,
+) -> RunsView:
     reference_time = now or datetime.now(UTC)
     resolved_log_dir = Path(log_dir)
+    sanitized_page_size = max(1, completed_page_size)
     with database.connection_context():
         execution_primary_key = getattr(JobExecution, "_meta").primary_key
         jobs = tuple(Job.select(Job, Source).join(Source).order_by(Source.name.asc()))
@@ -668,7 +686,7 @@ def load_runs_view(
             .where(JobExecution.running_status == JobExecutionStatus.RUNNING)
             .order_by(JobExecution.started_at.desc())
         )
-        completed_executions = tuple(
+        completed_query = (
             JobExecution.select(JobExecution, Job, Source)
             .join(Job)
             .join(Source)
@@ -682,7 +700,14 @@ def load_runs_view(
                 )
             )
             .order_by(JobExecution.ended_at.desc())
-            .limit(20)
+        )
+        completed_total_count = completed_query.count()
+        completed_total_pages = max(
+            1, math.ceil(completed_total_count / sanitized_page_size)
+        )
+        sanitized_completed_page = min(max(1, completed_page), completed_total_pages)
+        completed_executions = tuple(
+            completed_query.paginate(sanitized_completed_page, sanitized_page_size)
         )
 
         running_by_job = {
@@ -725,9 +750,49 @@ def load_runs_view(
             _project_completed_execution(execution, resolved_log_dir, reference_time)
             for execution in completed_executions
         ),
+        "completed_page": sanitized_completed_page,
+        "completed_page_size": sanitized_page_size,
+        "completed_total_count": completed_total_count,
+        "completed_total_pages": completed_total_pages,
     }
 
 
+def clear_completed_executions(*, log_dir: str | Path) -> int:
+    resolved_log_dir = Path(log_dir)
+    with database.connection_context():
+        execution_primary_key = getattr(JobExecution, "_meta").primary_key
+        completed_executions = tuple(
+            JobExecution.select(JobExecution, Job)
+            .join(Job)
+            .where(
+                JobExecution.running_status.in_(
+                    (
+                        JobExecutionStatus.SUCCEEDED,
+                        JobExecutionStatus.FAILED,
+                        JobExecutionStatus.CANCELED,
+                    )
+                )
+            )
+        )
+        if not completed_executions:
+            return 0
+
+        for execution in completed_executions:
+            job = cast(Job, execution.job)
+            prefix = f"job-{_job_id(job)}-execution-{_execution_id(execution)}"
+            for artifact_path in resolved_log_dir.glob(f"{prefix}.*"):
+                artifact_path.unlink(missing_ok=True)
+
+        execution_ids = tuple(
+            _execution_id(execution) for execution in completed_executions
+        )
+        return (
+            JobExecution.delete()
+            .where(execution_primary_key.in_(execution_ids))
+            .execute()
+        )
+
+
 def load_dashboard_view(
     *, log_dir: str | Path, now: datetime | None = None
 ) -> dict[str, object]: