providers.py 2.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364
  1. from __future__ import annotations
  2. from typing import Any, Protocol
  3. from .repositories import ScriptBuildRepository
  4. class RuntimeDataProvider(Protocol):
  5. mode: str
  6. def list_builds(
  7. self, *, limit: int = 30, status: str | None = None
  8. ) -> list[dict[str, Any]]: ...
  9. def load_bundle(self, script_build_id: int) -> dict[str, Any]: ...
  10. def load_current_artifact(self, script_build_id: int) -> dict[str, Any]: ...
  11. def load_event_detail(
  12. self, script_build_id: int, event_id: int
  13. ) -> dict[str, Any]: ...
  14. def load_event_details(
  15. self, script_build_id: int, event_ids: list[int]
  16. ) -> dict[int, dict[str, Any]]: ...
  17. def load_prompt_context(
  18. self, script_build_id: int, prompt_ref: str
  19. ) -> dict[str, Any]: ...
  20. class LocalDatabaseProvider:
  21. """V8 production provider: direct, visualization-only, read-only DB access."""
  22. mode = "local-db-read-only"
  23. def __init__(self, repository: ScriptBuildRepository | None = None):
  24. self.repository = repository or ScriptBuildRepository()
  25. def list_builds(
  26. self, *, limit: int = 30, status: str | None = None
  27. ) -> list[dict[str, Any]]:
  28. return self.repository.list_builds(limit=limit, status=status)
  29. def load_bundle(self, script_build_id: int) -> dict[str, Any]:
  30. return self.repository.load_bundle(script_build_id)
  31. def load_current_artifact(self, script_build_id: int) -> dict[str, Any]:
  32. return self.repository.load_current_artifact(script_build_id)
  33. def load_event_detail(
  34. self, script_build_id: int, event_id: int
  35. ) -> dict[str, Any]:
  36. return self.repository.load_event_detail(script_build_id, event_id)
  37. def load_event_details(
  38. self, script_build_id: int, event_ids: list[int]
  39. ) -> dict[int, dict[str, Any]]:
  40. return self.repository.load_event_details(script_build_id, event_ids)
  41. def load_prompt_context(
  42. self, script_build_id: int, prompt_ref: str
  43. ) -> dict[str, Any]:
  44. return self.repository.load_prompt_context(script_build_id, prompt_ref)