151 lines
4.4 KiB
Python
151 lines
4.4 KiB
Python
from datetime import datetime, timezone
|
|
from typing import Any
|
|
from uuid import UUID, uuid4
|
|
|
|
from app.infra.db.postgresql.analysis import AnalysisRepository
|
|
from app.native.wndb.core.connection import project_connection, project_transaction
|
|
from app.infra.db.project_routing import get_active_project_routing
|
|
from app.domain.time import parse_utc_time
|
|
|
|
|
|
def store_scheme_info(
|
|
name: str,
|
|
scheme_name: str,
|
|
scheme_type: str,
|
|
username: str,
|
|
scheme_start_time: datetime | str,
|
|
scheme_detail: dict[str, Any],
|
|
) -> UUID:
|
|
"""Create one completed analysis run; its name remains a display label."""
|
|
return create_analysis_run(
|
|
name=name,
|
|
scheme_name=scheme_name,
|
|
scheme_type=scheme_type,
|
|
username=username,
|
|
scheme_start_time=scheme_start_time,
|
|
scheme_detail=scheme_detail,
|
|
status="completed",
|
|
)
|
|
|
|
|
|
def create_analysis_run(
|
|
name: str,
|
|
scheme_name: str,
|
|
scheme_type: str,
|
|
username: str,
|
|
scheme_start_time: datetime | str,
|
|
scheme_detail: dict[str, Any],
|
|
*,
|
|
status: str = "running",
|
|
) -> UUID:
|
|
started_at = parse_utc_time(scheme_start_time, field_name="scheme_start_time")
|
|
run_id = uuid4()
|
|
with project_connection(name) as conn:
|
|
with conn.transaction():
|
|
AnalysisRepository.create_run_sync(
|
|
conn,
|
|
run_id=run_id,
|
|
name=scheme_name,
|
|
run_type=scheme_type,
|
|
created_by=username,
|
|
started_at=started_at,
|
|
status=status,
|
|
parameters=scheme_detail,
|
|
)
|
|
_append_run_event(
|
|
conn, run_id, revision=1, status=status,
|
|
occurred_at=started_at, results_committed=False,
|
|
)
|
|
return run_id
|
|
|
|
|
|
def store_analysis_run_with_result(
|
|
*,
|
|
name: str,
|
|
scheme_name: str,
|
|
scheme_type: str,
|
|
username: str,
|
|
scheme_start_time: datetime | str,
|
|
scheme_detail: dict[str, Any],
|
|
result_type: str,
|
|
result_payload: dict[str, Any],
|
|
) -> UUID:
|
|
"""Atomically persist one BizDB run and its non-timeseries result."""
|
|
started_at = parse_utc_time(scheme_start_time, field_name="scheme_start_time")
|
|
run_id = uuid4()
|
|
with project_transaction(name) as conn:
|
|
AnalysisRepository.create_run_sync(
|
|
conn,
|
|
run_id=run_id,
|
|
name=scheme_name,
|
|
run_type=scheme_type,
|
|
created_by=username,
|
|
started_at=started_at,
|
|
status="completed",
|
|
parameters=scheme_detail,
|
|
)
|
|
AnalysisRepository.insert_result_sync(
|
|
conn,
|
|
run_id,
|
|
result_type=result_type,
|
|
payload=result_payload,
|
|
)
|
|
_append_run_event(
|
|
conn, run_id, revision=1, status="completed",
|
|
occurred_at=started_at, results_committed=True,
|
|
)
|
|
return run_id
|
|
|
|
|
|
def update_analysis_run(
|
|
name: str,
|
|
run_id: UUID,
|
|
*,
|
|
status: str,
|
|
username: str,
|
|
scheme_detail: dict[str, Any],
|
|
) -> None:
|
|
with project_connection(name) as conn:
|
|
with conn.transaction():
|
|
revision = AnalysisRepository.update_run_sync(
|
|
conn,
|
|
run_id,
|
|
status=status,
|
|
created_by=username,
|
|
parameters=scheme_detail,
|
|
)
|
|
_append_run_event(
|
|
conn, run_id, revision=revision, status=status,
|
|
occurred_at=datetime.now(timezone.utc),
|
|
results_committed=status == "completed",
|
|
)
|
|
|
|
|
|
def _append_run_event(
|
|
conn, run_id: UUID, *, revision: int, status: str,
|
|
occurred_at: datetime, results_committed: bool,
|
|
) -> None:
|
|
routing = get_active_project_routing()
|
|
if routing is None or routing.project_id is None:
|
|
return
|
|
AnalysisRepository.append_run_changed_event_sync(
|
|
conn,
|
|
project_id=routing.project_id,
|
|
run_id=run_id,
|
|
revision=revision,
|
|
status=status,
|
|
occurred_at=occurred_at,
|
|
results_committed=results_committed,
|
|
)
|
|
|
|
|
|
def get_analysis_run(name: str, run_id: UUID) -> dict[str, Any]:
|
|
with project_connection(name) as conn:
|
|
row = AnalysisRepository.get_run_sync(conn, run_id)
|
|
if row is None:
|
|
return {}
|
|
parameters = row.get("parameters")
|
|
return dict(row) | {
|
|
"parameters": parameters if isinstance(parameters, dict) else {}
|
|
}
|