# #region ValidationService [C:4] [TYPE Module] [SEMANTICS validation, service, sqlalchemy, task] # @BRIEF Business logic for validation task CRUD and trigger-run — v2 with source-aware policy management. # @LAYER Service # @RELATION DEPENDS_ON -> [LLMProviderService] # @RELATION DEPENDS_ON -> [SupersetContextExtractor] # @RELATION DEPENDS_ON -> [ValidationPolicy] # @RELATION DEPENDS_ON -> [ValidationRun] # @RELATION DEPENDS_ON -> [ValidationSource] from datetime import time from typing import Any from sqlalchemy.orm import Session from ..core.logger import logger from ..core.task_manager.manager import TaskManager from ..models.llm import LLMProvider, ValidationPolicy, ValidationRecord, ValidationRun, ValidationSource from ..schemas.validation import ( SourceInput, SourceResponse, ValidationTaskCreate, ValidationTaskResponse, ValidationTaskUpdate, ) # #region _extract_screenshot_paths [C:2] [TYPE Function] # @BRIEF Extract screenshot_paths from raw_response JSON when the column is empty (backward compat). def _extract_screenshot_paths(raw_response): if not raw_response: return [] import json try: parsed = json.loads(raw_response) paths = parsed.get("screenshot_paths") or [] if isinstance(paths, list): return paths except (json.JSONDecodeError, TypeError): pass return [] # #endregion _extract_screenshot_paths from .llm_provider import LLMProviderService def _parse_time(value: str | None) -> time | None: """Parse HH:MM string to datetime.time, or return None.""" if not value: return None try: parts = value.strip().split(":") return time(int(parts[0]), int(parts[1])) except (ValueError, IndexError): return None def _format_time(value: time | None) -> str | None: """Format datetime.time to HH:MM string, or return None.""" if value is None: return None return f"{value.hour:02d}:{value.minute:02d}" # #region ValidationTaskService [C:4] [TYPE Class] # @BRIEF Validation task (policy) CRUD with provider validation and trigger-run. # @RATIONALE Service-layer class for validation task CRUD with provider validation and trigger-run. # Separates API layer concerns from business logic; enables unit testing without HTTP. # @REJECTED Embedding CRUD in route handlers rejected — would duplicate validation logic across # endpoints; centralized service ensures consistent provider/environment validation # before persistence. class ValidationTaskService: def __init__(self, db: Session, config_manager=None, username: str | None = None): self.db = db self.config_manager = config_manager self.username = username or "system" # #region _validate_provider [C:2] [TYPE Function] # @RATIONALE Centralizes LLM provider check with configurable multimodal requirement. Path A # (screenshot_enabled=True) requires multimodal; Path B (screenshot_enabled=False) # only checks existence and is_active. # @REJECTED Duplicating multimodal check in each caller rejected — would create maintenance # burden if provider validation rules change or error messages need updating. def _validate_provider(self, provider_id: str, require_multimodal: bool = True) -> LLMProvider: if not provider_id or not provider_id.strip(): raise ValueError("provider_id is required") svc = LLMProviderService(self.db) provider = svc.get_provider(provider_id) if not provider: raise ValueError(f"LLM provider '{provider_id}' not found") if not bool(provider.is_active): raise ValueError( f"LLM provider '{provider.name}' is not active. " "Only active providers can be used for validation." ) if require_multimodal and not bool(provider.is_multimodal): raise ValueError( f"LLM provider '{provider.name}' is not multimodal. " "Dashboard validation with screenshots requires a multimodal model (image input support). " "Set screenshot_enabled=False to use a text-only provider." ) return provider # #endregion _validate_provider # #region _validate_environment [C:1] [TYPE Function] def _validate_environment(self, environment_id: str) -> None: if not self.config_manager: return env = self.config_manager.get_environment(environment_id) if not env: raise ValueError(f"Environment '{environment_id}' not found") # #endregion _validate_environment # #region _parse_superset_link [C:3] [TYPE Function] # @RATIONALE Uses SupersetContextExtractor to validate and extract context from a full Superset # dashboard URL. Called during source creation when type == "dashboard_url". # @REJECTED Silently accepting unparseable URLs rejected — would create broken sources that # fail at runtime during validation without clear error attribution. def _parse_superset_link(self, url: str, environment_id: str) -> dict: if not self.config_manager: raise ValueError("Config manager not available — cannot validate Superset URL") env = self.config_manager.get_environment(environment_id) if not env: raise ValueError(f"Environment '{environment_id}' not found — cannot validate Superset URL") from ..core.utils.superset_context_extractor import SupersetContextExtractor extractor = SupersetContextExtractor(environment=env) parsed = extractor.parse_superset_link(url) return { "source_url": parsed.source_url, "dataset_ref": parsed.dataset_ref, "dataset_id": parsed.dataset_id, "dashboard_id": parsed.dashboard_id, "chart_id": parsed.chart_id, "resource_type": parsed.resource_type, "partial_recovery": parsed.partial_recovery, "unresolved_references": list(parsed.unresolved_references), } # #endregion _parse_superset_link # #region _resolve_dashboard_title [C:2] [TYPE Function] # @BRIEF Fetch a dashboard title from the Superset API by its numeric ID. # @SIDE_EFFECT Makes an HTTP GET to Superset REST API via the config-manager. # @RETURNS The dashboard title string, or None if it cannot be resolved. def _resolve_dashboard_title(self, environment_id: str, dashboard_id: str) -> str | None: try: from ..core.superset_client import SupersetClient if not self.config_manager: return None env = self.config_manager.get_environment(environment_id) if not env: return None client = SupersetClient(env) resp = client.get_dashboard(dashboard_id) # Superset API returns { "result": { "dashboard_title": "..." } } result = resp.get("result", resp) title = result.get("dashboard_title") or result.get("title") return str(title) if title else None except Exception: return None # #endregion _resolve_dashboard_title # #region _resolve_sources [C:3] [TYPE Function] # @RATIONALE Resolves SourceInput list into ValidationSource ORM objects. Dashboard_url sources # are parsed via SupersetContextExtractor for validation and context extraction. # Falls back to dashboard_ids for backward compatibility. # @REJECTED Mixing source types in a single method that also persists rejected — separation # from DB commit gives callers control over transaction boundaries. def _resolve_sources( self, sources_input: list[SourceInput] | None, dashboard_ids: list[str] | None, policy_id: str, environment_id: str, ) -> list[ValidationSource]: sources: list[ValidationSource] = [] # Priority: explicit sources list (v2) if sources_input: for s in sources_input: parsed_context = None status = "valid" resolved_dashboard_id = None title = None if s.type == "dashboard_url": parsed_context = self._parse_superset_link(s.value, environment_id) if parsed_context.get("dashboard_id"): resolved_dashboard_id = str(parsed_context["dashboard_id"]) if parsed_context.get("partial_recovery"): status = "partial_recovery" elif s.type == "dashboard_id": resolved_dashboard_id = s.value # Resolve dashboard title from Superset API if resolved_dashboard_id and not title: title = self._resolve_dashboard_title(environment_id, resolved_dashboard_id) sources.append(ValidationSource( policy_id=policy_id, type=s.type, value=s.value, parsed_context=parsed_context, status=status, resolved_dashboard_id=resolved_dashboard_id, title=title, )) # Fallback: dashboard_ids (v1 backward compat) elif dashboard_ids: for did in dashboard_ids: title = self._resolve_dashboard_title(environment_id, str(did)) sources.append(ValidationSource( policy_id=policy_id, type="dashboard_id", value=str(did), status="valid", resolved_dashboard_id=str(did), title=title, )) return sources # #endregion _resolve_sources # #region _policy_to_response [C:3] [TYPE Function] # @RATIONALE Convert ValidationPolicy ORM to response DTO using the ValidationRun model for # last run info instead of individual ValidationRecords, giving correct aggregate # status for v2 multi-dashboard runs. # @REJECTED Filtering by dashboard_id+environment_id for last run rejected — with v2 multi-source # policies, runs are policy-scoped, not per-dashboard. def _policy_to_response(self, policy: ValidationPolicy) -> ValidationTaskResponse: # Build sources response sources_list: list[SourceResponse] = [] if policy.sources: for s in policy.sources: sources_list.append(SourceResponse( id=s.id, type=s.type, value=s.value, status=s.status, title=s.title, resolved_dashboard_id=s.resolved_dashboard_id, last_error=s.last_error, last_checked_at=s.last_checked_at, created_at=s.created_at, )) # Last run via ValidationRun (v2 aggregate) last_run = ( self.db.query(ValidationRun) .filter(ValidationRun.policy_id == policy.id) .order_by(ValidationRun.started_at.desc()) .first() ) return ValidationTaskResponse( id=policy.id, name=policy.name, description=policy.description, environment_id=policy.environment_id, is_active=bool(policy.is_active) if policy.is_active is not None else True, dashboard_ids=list(policy.dashboard_ids) if policy.dashboard_ids else [], provider_id=policy.provider_id, schedule_days=list(policy.schedule_days) if policy.schedule_days else None, window_start=_format_time(policy.window_start), window_end=_format_time(policy.window_end), notify_owners=bool(policy.notify_owners) if policy.notify_owners is not None else True, custom_channels=list(policy.custom_channels) if policy.custom_channels else None, alert_condition=policy.alert_condition or "FAIL_ONLY", created_at=policy.created_at, updated_at=policy.updated_at, last_run_status=str(last_run.status) if last_run else None, last_run_at=last_run.started_at if last_run else None, last_run_summary={ "pass": last_run.pass_count or 0, "warn": last_run.warn_count or 0, "fail": last_run.fail_count or 0, } if last_run else None, # v2 fields screenshot_enabled=bool(policy.screenshot_enabled) if policy.screenshot_enabled is not None else True, logs_enabled=bool(policy.logs_enabled) if policy.logs_enabled is not None else True, execute_chart_data=bool(policy.execute_chart_data) if policy.execute_chart_data is not None else False, llm_batch_size=policy.llm_batch_size or 1, policy_dashboard_concurrency_limit=policy.policy_dashboard_concurrency_limit or 3, prompt_template=policy.prompt_template, sources=sources_list, ) # #endregion _policy_to_response # #region _run_to_dict [C:2] [TYPE Function] # @BRIEF Convert ValidationRun ORM object to a serializable dict for run history APIs. # Computes duration_ms from started_at/finished_at for frontend display. # @RATIONALE duration_ms added so the frontend can show "Длительность: 3m 12s" on the # run detail page. Previously duration was always "-" even for completed runs. # @POST Returns dict with duration_ms = int (ms) if both started_at and finished_at # are present, else None. All other fields are direct ORM mappings. def _run_to_dict(self, run: ValidationRun) -> dict: from datetime import timezone as _tz duration_ms: int | None = None if run.started_at and run.finished_at: started = run.started_at.replace(tzinfo=_tz.utc) if run.started_at.tzinfo is None else run.started_at finished = run.finished_at.replace(tzinfo=_tz.utc) if run.finished_at.tzinfo is None else run.finished_at duration_ms = int((finished - started).total_seconds() * 1000) return { "id": run.id, "policy_id": run.policy_id, "task_id": run.task_id, "status": run.status, "trigger": run.trigger, "started_at": run.started_at.isoformat() if run.started_at else None, "finished_at": run.finished_at.isoformat() if run.finished_at else None, "duration_ms": duration_ms, "dashboard_count": run.dashboard_count or 0, "pass_count": run.pass_count or 0, "warn_count": run.warn_count or 0, "fail_count": run.fail_count or 0, "unknown_count": run.unknown_count or 0, "created_at": run.created_at.isoformat() if run.created_at else None, } # #endregion _run_to_dict # #region _record_to_dict [C:2] [TYPE Function] # @BRIEF Convert ValidationRecord ORM object to a serializable dict with all v2 fields. def _record_to_dict(self, record: ValidationRecord) -> dict: # Try to resolve dashboard title from sources dashboard_title = record.dashboard_id try: source = ( self.db.query(ValidationSource) .filter(ValidationSource.resolved_dashboard_id == record.dashboard_id) .first() ) if source and source.title: dashboard_title = source.title elif source and record.environment_id: # Fallback: resolve from Superset API and cache back to source resolved = self._resolve_dashboard_title(record.environment_id, record.dashboard_id) if resolved: dashboard_title = resolved source.title = resolved self.db.flush() except Exception: pass return { "id": record.id, "run_id": record.run_id, "policy_id": record.policy_id, "dashboard_id": record.dashboard_id, "dashboard_title": dashboard_title, "environment_id": record.environment_id, "status": record.status, "summary": record.summary, "issues": record.issues or [], "raw_response": record.raw_response, "screenshot_path": record.screenshot_path, # screenshot_paths column may be empty for runs created before the fix; # fall back to extracting from raw_response JSON "screenshot_paths": record.screenshot_paths or _extract_screenshot_paths(record.raw_response), "execution_path": record.execution_path, "dataset_health": record.dataset_health, "chart_data_results": record.chart_data_results or [], "tab_screenshots": record.tab_screenshots or [], "logs_sent_to_llm": record.logs_sent_to_llm or [], "token_usage": record.token_usage, "timings": record.timings, "timestamp": record.timestamp.isoformat() if record.timestamp else None, } # #endregion _record_to_dict # #region list_tasks [C:3] [TYPE Function] # @RATIONALE Uses query-time filtering with optional is_active/environment_id params rather # than post-filtering results in memory, reducing overhead with large task counts. # @REJECTED In-memory filtering after loading all rows rejected — would be inefficient with # hundreds of tasks; database-level filtering scales better. def list_tasks( self, is_active: bool | None = None, environment_id: str | None = None, search: str | None = None, page: int = 1, page_size: int = 20, ) -> tuple[int, list[ValidationTaskResponse]]: query = self.db.query(ValidationPolicy) if is_active is not None: query = query.filter(ValidationPolicy.is_active == is_active) if environment_id: query = query.filter(ValidationPolicy.environment_id == environment_id) if search: query = query.filter(ValidationPolicy.name.ilike(f"%{search}%")) total = query.count() policies = ( query.order_by(ValidationPolicy.created_at.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) return total, [self._policy_to_response(p) for p in policies] # #endregion list_tasks # #region create_task [C:4] [TYPE Function] # @RATIONALE Validates provider (multimodal check depends on screenshot_enabled flag) and # environment before persisting the policy. Resolves sources from explicit input # or falls back to dashboard_ids for backward compatibility. # @REJECTED Lazy validation on first run rejected — would allow creation of invalid task # configurations with incorrect providers that fail only at runtime. def create_task(self, payload: ValidationTaskCreate) -> ValidationTaskResponse: self._validate_provider(payload.provider_id, require_multimodal=payload.screenshot_enabled) self._validate_environment(payload.environment_id) ws = _parse_time(payload.window_start) we = _parse_time(payload.window_end) policy = ValidationPolicy( name=payload.name, description=payload.description, environment_id=payload.environment_id, is_active=payload.is_active, dashboard_ids=payload.dashboard_ids or [], provider_id=payload.provider_id, schedule_days=payload.schedule_days or [], window_start=ws or time(9, 0), window_end=we or time(18, 0), notify_owners=payload.notify_owners, custom_channels=payload.custom_channels, alert_condition=payload.alert_condition or "FAIL_ONLY", # v2 fields screenshot_enabled=payload.screenshot_enabled, logs_enabled=payload.logs_enabled, execute_chart_data=payload.execute_chart_data, llm_batch_size=payload.llm_batch_size, policy_dashboard_concurrency_limit=payload.policy_dashboard_concurrency_limit, prompt_template=payload.prompt_template, ) self.db.add(policy) self.db.flush() # acquire policy.id for source foreign keys # Resolve sources sources = self._resolve_sources( sources_input=payload.sources, dashboard_ids=payload.dashboard_ids, policy_id=policy.id, environment_id=payload.environment_id, ) for source in sources: self.db.add(source) self.db.commit() self.db.refresh(policy) logger.info( "[ValidationTaskService] Created task '%s' (id=%s) with %d sources", policy.name, policy.id, len(sources), ) return self._policy_to_response(policy) # #endregion create_task # #region get_task [C:3] [TYPE Function] # @RATIONALE Returns full task response including last run status via ValidationRun — gives # the UI immediate visibility into task health without a second API call. # @REJECTED Separate query for last run status rejected — double-query pattern would add # latency on every task detail request. def get_task(self, task_id: str) -> ValidationTaskResponse: policy = self.db.query(ValidationPolicy).filter(ValidationPolicy.id == task_id).first() if not policy: raise ValueError(f"Validation task '{task_id}' not found") return self._policy_to_response(policy) # #endregion get_task # #region _apply_time_field [C:2] [TYPE Function] # @BRIEF Parse and apply a time window field from update data, if present. def _apply_time_field(self, update_data: dict, policy: ValidationPolicy, field: str) -> None: if field in update_data: parsed = _parse_time(update_data.pop(field, None)) if parsed is not None: setattr(policy, field, parsed) # #endregion _apply_time_field # #region _apply_sources_update [C:2] [TYPE Function] # @BRIEF Replace all ValidationSource rows for a policy when sources are provided in update. def _apply_sources_update(self, update_data: dict, policy: ValidationPolicy) -> None: if "sources" not in update_data: return sources_data = update_data.pop("sources") if sources_data is None: return self.db.query(ValidationSource).filter( ValidationSource.policy_id == policy.id ).delete(synchronize_session=False) new_sources = self._resolve_sources( sources_input=sources_data, dashboard_ids=None, policy_id=policy.id, environment_id=update_data.get("environment_id", policy.environment_id), ) for s in new_sources: self.db.add(s) # #endregion _apply_sources_update # #region update_task [C:3] [TYPE Function] # @RATIONALE Re-validates provider/environment only when those fields change (via exclude_unset), # avoiding unnecessary DB calls on partial updates. Delegates time parsing and source # replacement to extracted methods to stay within CC ≤ 10. # @REJECTED Full re-validation on every update rejected — would prevent updating non-critical # fields (e.g., schedule, name) when a provider is temporarily unavailable. def update_task(self, task_id: str, payload: ValidationTaskUpdate) -> ValidationTaskResponse: policy = self.db.query(ValidationPolicy).filter(ValidationPolicy.id == task_id).first() if not policy: raise ValueError(f"Validation task '{task_id}' not found") update_data = payload.model_dump(exclude_unset=True) if update_data.get("provider_id"): require_modal = update_data.get("screenshot_enabled", policy.screenshot_enabled) self._validate_provider(update_data["provider_id"], require_multimodal=bool(require_modal)) if update_data.get("environment_id"): self._validate_environment(update_data["environment_id"]) self._apply_time_field(update_data, policy, "window_start") self._apply_time_field(update_data, policy, "window_end") self._apply_sources_update(update_data, policy) for key, value in update_data.items(): if hasattr(policy, key): setattr(policy, key, value) self.db.commit() self.db.refresh(policy) logger.info("[ValidationTaskService] Updated task '%s' (id=%s)", policy.name, policy.id) return self._policy_to_response(policy) # #endregion update_task # #region delete_task [C:3] [TYPE Function] # @RATIONALE Sources are cascade-deleted via SQLAlchemy relationship (cascade="all, delete-orphan"). # When delete_runs=True, also removes ValidationRun and their ValidationRecord children. # @REJECTED Always-cascade delete rejected — would destroy run history unexpectedly. Explicit # delete_runs flag with policy_id scoping gives users safe control. def delete_task(self, task_id: str, delete_runs: bool = False) -> None: policy = self.db.query(ValidationPolicy).filter(ValidationPolicy.id == task_id).first() if not policy: raise ValueError(f"Validation task '{task_id}' not found") if delete_runs: # Collect run IDs for this policy run_id_results = ( self.db.query(ValidationRun.id) .filter(ValidationRun.policy_id == policy.id) .all() ) run_ids = [r[0] for r in run_id_results] if run_ids: # Delete records belonging to these runs self.db.query(ValidationRecord).filter( ValidationRecord.run_id.in_(run_ids) ).delete(synchronize_session=False) # Delete runs themselves self.db.query(ValidationRun).filter( ValidationRun.policy_id == policy.id ).delete(synchronize_session=False) # Sources are cascade-deleted via relationship; delete the policy self.db.delete(policy) self.db.commit() logger.info( "[ValidationTaskService] Deleted task '%s' (id=%s) delete_runs=%s", policy.name, policy.id, delete_runs, ) # #endregion delete_task # #region trigger_run [C:4] [TYPE Function] # @RATIONALE Creates a ValidationRun record as an aggregate execution container. Checks FR-054 # (no concurrent running runs for the same policy) and global_validation_worker_limit # before spawning. Passes all resolved dashboard sources to the task manager plugin. # @REJECTED Triggering without ValidationRun creation rejected — runs need a persistent tracking # record for aggregate status, progress monitoring, and run history (FR-052). async def trigger_run(self, task_id: str, task_manager: TaskManager) -> dict: policy = self.db.query(ValidationPolicy).filter(ValidationPolicy.id == task_id).first() if not policy: raise ValueError(f"Validation task '{task_id}' not found") # FR-054: Reject if a running run already exists for this policy existing_run = ( self.db.query(ValidationRun) .filter( ValidationRun.policy_id == policy.id, ValidationRun.status == "running", ) .first() ) if existing_run: raise ValueError( f"Validation task '{task_id}' already has a running run (id={existing_run.id}). " "Wait for the current run to finish before triggering a new one." ) # Check global validation worker limit max_running = getattr( self.config_manager.config.settings, 'global_validation_worker_limit', 5 ) if self.config_manager else 5 running_count = ( self.db.query(ValidationRun) .filter(ValidationRun.status == "running") .count() ) if running_count >= max_running: raise ValueError( f"Global validation worker limit reached ({running_count}/{max_running}). " "Wait for running validations to complete before triggering new ones." ) # Get sources (v2) or fall back to dashboard_ids (v1) sources = ( self.db.query(ValidationSource) .filter(ValidationSource.policy_id == policy.id) .all() ) resolved_dashboard_ids: list[str] = [] if sources: resolved_dashboard_ids = [ s.resolved_dashboard_id or s.value for s in sources ] elif policy.dashboard_ids: resolved_dashboard_ids = list(policy.dashboard_ids) if not resolved_dashboard_ids: raise ValueError( f"Validation task '{task_id}' has no dashboard IDs or sources configured" ) # Validate provider with correct multimodal requirement if policy.provider_id: self._validate_provider( policy.provider_id, require_multimodal=bool(policy.screenshot_enabled), ) # Create ValidationRun record run = ValidationRun( policy_id=policy.id, trigger="manual", status="running", dashboard_count=len(resolved_dashboard_ids), ) self.db.add(run) self.db.flush() # Build task manager params including all v2 fields params: dict[str, Any] = { "dashboard_ids": resolved_dashboard_ids, "environment_id": policy.environment_id, "provider_id": policy.provider_id or "", "policy_id": task_id, "run_id": run.id, "screenshot_enabled": bool(policy.screenshot_enabled), "logs_enabled": bool(policy.logs_enabled), "execute_chart_data": bool(policy.execute_chart_data), "llm_batch_size": policy.llm_batch_size or 1, "concurrency_limit": policy.policy_dashboard_concurrency_limit or 3, "prompt_template": policy.prompt_template, } task = await task_manager.create_task( plugin_id="llm_dashboard_validation", params=params, ) # Link task ID back to the run record run.task_id = task.id self.db.commit() logger.info( "[ValidationTaskService] Triggered run for task '%s' run=%s spawned=%s dashboards=%d", policy.name, run.id, task.id, len(resolved_dashboard_ids), ) return {"task_id": task.id, "run_id": run.id} # #endregion trigger_run # #region get_run_history [C:3] [TYPE Function] # @BRIEF Return paginated ValidationRun list for a policy, ordered by started_at desc. # @RATIONALE Uses ValidationRun (aggregate per-run) rather than individual ValidationRecords, # giving a correct v2 view of execution history with aggregate pass/fail counts. # @REJECTED Using v1 ValidationRecord for run history rejected — v2 multi-dashboard runs # would produce multiple records per run, making the history view noisy. def get_run_history(self, policy_id: str, page: int = 1, page_size: int = 20) -> tuple[int, list[dict]]: query = self.db.query(ValidationRun).filter(ValidationRun.policy_id == policy_id) total = query.count() runs = ( query.order_by(ValidationRun.started_at.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) return total, [self._run_to_dict(r) for r in runs] # #endregion get_run_history # #region list_all_runs [C:2] [TYPE Function] # @BRIEF List runs across ALL tasks with optional filters (replaces v1 cross-task endpoint). def list_all_runs( self, status: str | None = None, environment_id: str | None = None, dashboard_id: str | None = None, page: int = 1, page_size: int = 20, ) -> tuple[int, list[dict]]: query = self.db.query(ValidationRecord) if status: query = query.filter(ValidationRecord.status == status.upper()) if environment_id: query = query.filter(ValidationRecord.environment_id == environment_id) if dashboard_id: query = query.filter(ValidationRecord.dashboard_id == dashboard_id) total = query.count() records = ( query.order_by(ValidationRecord.timestamp.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) return total, [self._record_to_dict(r) for r in records] # #endregion list_all_runs # #region get_run_detail [C:3] [TYPE Function] # @BRIEF Return full ValidationRun with all its ValidationRecords, including v2 metadata # (token_usage, timings, execution_path, dataset_health, etc.). # @RATIONALE Loads records with eager filtering on run_id for a complete picture of a single # validation execution across all its dashboards. # @REJECTED Embedding record data inside the run response without pagination rejected — for # very large runs (100+ dashboards), but acceptable for typical task sizes (<20). def get_run_detail(self, run_id: str) -> dict: run = self.db.query(ValidationRun).filter(ValidationRun.id == run_id).first() if not run: raise ValueError(f"Validation run '{run_id}' not found") records = ( self.db.query(ValidationRecord) .filter(ValidationRecord.run_id == run_id) .order_by(ValidationRecord.timestamp.desc()) .all() ) return { "run": self._run_to_dict(run), "records": [self._record_to_dict(r) for r in records], } # #endregion get_run_detail # #endregion ValidationTaskService # #endregion ValidationService