jiang
5966d039de
Reorganize algorithm packages by business responsibility, move orchestration into services, and keep database access behind pooled repositories. Harden analysis API validation, remove unsafe legacy simulation endpoints, and add regression and architecture boundary coverage. BREAKING CHANGE: legacy algorithm module paths and obsolete simulation endpoints are removed.
117 lines
3.1 KiB
Python
117 lines
3.1 KiB
Python
from datetime import datetime
|
|
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.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:
|
|
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,
|
|
)
|
|
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,
|
|
)
|
|
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:
|
|
AnalysisRepository.update_run_sync(
|
|
conn,
|
|
run_id,
|
|
status=status,
|
|
created_by=username,
|
|
parameters=scheme_detail,
|
|
)
|
|
|
|
|
|
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 {}
|
|
}
|