|
|
@@ -1,9 +1,10 @@
|
|
|
from __future__ import annotations
|
|
|
|
|
|
+import asyncio
|
|
|
import json
|
|
|
from typing import Any
|
|
|
|
|
|
-from sqlalchemy import or_, select
|
|
|
+from sqlalchemy import select
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
|
|
from script_build_host.domain.errors import InputRelationMismatch
|
|
|
@@ -44,14 +45,51 @@ def _json_value(value: Any) -> Any:
|
|
|
|
|
|
|
|
|
def _row_dict(row: Any, fields: tuple[str, ...]) -> dict[str, Any]:
|
|
|
- return {field: row[field] for field in fields}
|
|
|
+ return {field: row[field] if field in row else None for field in fields}
|
|
|
+
|
|
|
+
|
|
|
+async def _rows_by_primary_key(
|
|
|
+ sessions: async_sessionmaker[AsyncSession],
|
|
|
+ table: Any,
|
|
|
+ identifiers: list[int],
|
|
|
+ *,
|
|
|
+ columns: tuple[Any, ...] | None = None,
|
|
|
+) -> list[Any]:
|
|
|
+ """Read LOB-bearing legacy rows by PK; old RDS plans stall on covering-range lookups."""
|
|
|
+
|
|
|
+ limiter = asyncio.Semaphore(8)
|
|
|
+
|
|
|
+ async def load(identifier: int) -> Any:
|
|
|
+ async with limiter, sessions() as session:
|
|
|
+ return (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(*(columns or (table,))).where(table.c.id == identifier)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+
|
|
|
+ rows = []
|
|
|
+ for row in await asyncio.gather(*(load(identifier) for identifier in identifiers)):
|
|
|
+ if row is None:
|
|
|
+ raise InputRelationMismatch()
|
|
|
+ rows.append(row)
|
|
|
+ return rows
|
|
|
|
|
|
|
|
|
class LegacySqlAlchemyInputReader:
|
|
|
"""Anti-corruption adapter over the old tables; it never imports the old runtime package."""
|
|
|
|
|
|
- def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None:
|
|
|
+ def __init__(
|
|
|
+ self,
|
|
|
+ sessions: async_sessionmaker[AsyncSession],
|
|
|
+ *,
|
|
|
+ include_raw_upload: bool = False,
|
|
|
+ ) -> None:
|
|
|
self._sessions = sessions
|
|
|
+ self._include_raw_upload = include_raw_upload
|
|
|
|
|
|
async def read_topic_graph(
|
|
|
self, *, execution_id: int, topic_build_id: int, topic_id: int
|
|
|
@@ -71,10 +109,15 @@ class LegacySqlAlchemyInputReader:
|
|
|
.mappings()
|
|
|
.one_or_none()
|
|
|
)
|
|
|
- build = (
|
|
|
+ build_base = (
|
|
|
(
|
|
|
await session.execute(
|
|
|
- select(topic_build_record).where(
|
|
|
+ select(
|
|
|
+ topic_build_record.c.id,
|
|
|
+ topic_build_record.c.execution_id,
|
|
|
+ topic_build_record.c.status,
|
|
|
+ topic_build_record.c.origin,
|
|
|
+ ).where(
|
|
|
topic_build_record.c.id == topic_build_id,
|
|
|
topic_build_record.c.execution_id == execution_id,
|
|
|
topic_build_record.c.is_deleted.is_(False),
|
|
|
@@ -84,6 +127,26 @@ class LegacySqlAlchemyInputReader:
|
|
|
.mappings()
|
|
|
.one_or_none()
|
|
|
)
|
|
|
+ build = dict(build_base) if build_base is not None else None
|
|
|
+ if build is not None:
|
|
|
+ build_columns = (
|
|
|
+ "demand",
|
|
|
+ "demand_constraints",
|
|
|
+ "agent_type",
|
|
|
+ "agent_config",
|
|
|
+ "strategies_config",
|
|
|
+ "personal_config",
|
|
|
+ "start_time",
|
|
|
+ "end_time",
|
|
|
+ *(("raw_upload_data",) if self._include_raw_upload else ()),
|
|
|
+ )
|
|
|
+ for column_name in build_columns:
|
|
|
+ column = topic_build_record.c[column_name]
|
|
|
+ build[column_name] = (
|
|
|
+ await session.execute(
|
|
|
+ select(column).where(topic_build_record.c.id == topic_build_id)
|
|
|
+ )
|
|
|
+ ).scalar_one()
|
|
|
topic = (
|
|
|
(
|
|
|
await session.execute(
|
|
|
@@ -108,10 +171,10 @@ class LegacySqlAlchemyInputReader:
|
|
|
):
|
|
|
raise InputRelationMismatch()
|
|
|
|
|
|
- point_rows = (
|
|
|
+ point_id_rows = (
|
|
|
(
|
|
|
await session.execute(
|
|
|
- select(topic_build_point)
|
|
|
+ select(topic_build_point.c.id)
|
|
|
.where(
|
|
|
topic_build_point.c.topic_id == topic_id,
|
|
|
topic_build_point.c.build_id == topic_build_id,
|
|
|
@@ -120,13 +183,13 @@ class LegacySqlAlchemyInputReader:
|
|
|
.order_by(topic_build_point.c.id)
|
|
|
)
|
|
|
)
|
|
|
- .mappings()
|
|
|
+ .scalars()
|
|
|
.all()
|
|
|
)
|
|
|
- item_rows = (
|
|
|
+ item_id_rows = (
|
|
|
(
|
|
|
await session.execute(
|
|
|
- select(topic_build_composition_item)
|
|
|
+ select(topic_build_composition_item.c.id)
|
|
|
.where(
|
|
|
topic_build_composition_item.c.topic_id == topic_id,
|
|
|
topic_build_composition_item.c.build_id == topic_build_id,
|
|
|
@@ -138,58 +201,113 @@ class LegacySqlAlchemyInputReader:
|
|
|
)
|
|
|
)
|
|
|
)
|
|
|
- .mappings()
|
|
|
+ .scalars()
|
|
|
.all()
|
|
|
)
|
|
|
- point_ids = {int(row["id"]) for row in point_rows}
|
|
|
- item_ids = {int(row["id"]) for row in item_rows}
|
|
|
- point_relation_rows = (
|
|
|
- (
|
|
|
- await session.execute(
|
|
|
- select(topic_build_point_item_relation).where(
|
|
|
- or_(
|
|
|
+ point_id_order = [int(value) for value in point_id_rows]
|
|
|
+ item_id_order = [int(value) for value in item_id_rows]
|
|
|
+ point_rows = await _rows_by_primary_key(
|
|
|
+ self._sessions,
|
|
|
+ topic_build_point,
|
|
|
+ point_id_order,
|
|
|
+ columns=tuple(
|
|
|
+ topic_build_point.c[name]
|
|
|
+ for name in (
|
|
|
+ "id",
|
|
|
+ "topic_id",
|
|
|
+ "build_id",
|
|
|
+ "point_type",
|
|
|
+ "point_result",
|
|
|
+ "is_active",
|
|
|
+ "note",
|
|
|
+ )
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ item_rows = await _rows_by_primary_key(
|
|
|
+ self._sessions,
|
|
|
+ topic_build_composition_item,
|
|
|
+ item_id_order,
|
|
|
+ columns=tuple(
|
|
|
+ topic_build_composition_item.c[name]
|
|
|
+ for name in (
|
|
|
+ "id",
|
|
|
+ "topic_id",
|
|
|
+ "build_id",
|
|
|
+ "item_level",
|
|
|
+ "dimension",
|
|
|
+ "point_type",
|
|
|
+ "element_name",
|
|
|
+ "category_path",
|
|
|
+ "category_id",
|
|
|
+ "derivation_type",
|
|
|
+ "step",
|
|
|
+ "sort_order",
|
|
|
+ "is_active",
|
|
|
+ "note",
|
|
|
+ "created_at",
|
|
|
+ "updated_at",
|
|
|
+ )
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ point_ids = set(point_id_order)
|
|
|
+ item_ids = set(item_id_order)
|
|
|
+ point_relation_ids = (
|
|
|
+ list(
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(topic_build_point_item_relation.c.id).where(
|
|
|
topic_build_point_item_relation.c.point_id.in_(point_ids),
|
|
|
- topic_build_point_item_relation.c.item_id.in_(item_ids),
|
|
|
- ),
|
|
|
+ )
|
|
|
)
|
|
|
- )
|
|
|
+ ).scalars()
|
|
|
)
|
|
|
- .mappings()
|
|
|
- .all()
|
|
|
- if point_ids or item_ids
|
|
|
+ if point_ids
|
|
|
else []
|
|
|
)
|
|
|
- item_relation_rows = (
|
|
|
- (
|
|
|
- await session.execute(
|
|
|
- select(topic_build_item_relation)
|
|
|
- .where(
|
|
|
- topic_build_item_relation.c.topic_id == topic_id,
|
|
|
+ point_relation_rows = await _rows_by_primary_key(
|
|
|
+ self._sessions,
|
|
|
+ topic_build_point_item_relation,
|
|
|
+ [int(value) for value in point_relation_ids],
|
|
|
+ )
|
|
|
+ item_relation_ids = (
|
|
|
+ list(
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(topic_build_item_relation.c.id)
|
|
|
+ .where(
|
|
|
+ topic_build_item_relation.c.topic_id == topic_id,
|
|
|
+ )
|
|
|
+ .order_by(topic_build_item_relation.c.id)
|
|
|
)
|
|
|
- .order_by(topic_build_item_relation.c.id)
|
|
|
- )
|
|
|
+ ).scalars()
|
|
|
)
|
|
|
- .mappings()
|
|
|
- .all()
|
|
|
if item_ids
|
|
|
else []
|
|
|
)
|
|
|
- source_rows = (
|
|
|
- (
|
|
|
- await session.execute(
|
|
|
- select(topic_build_item_source)
|
|
|
- .where(
|
|
|
- topic_build_item_source.c.topic_id == topic_id,
|
|
|
- topic_build_item_source.c.is_active.is_(True),
|
|
|
+ item_relation_rows = await _rows_by_primary_key(
|
|
|
+ self._sessions,
|
|
|
+ topic_build_item_relation,
|
|
|
+ [int(value) for value in item_relation_ids],
|
|
|
+ )
|
|
|
+ source_ids = (
|
|
|
+ list(
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(topic_build_item_source.c.id)
|
|
|
+ .where(
|
|
|
+ topic_build_item_source.c.topic_id == topic_id,
|
|
|
+ topic_build_item_source.c.is_active.is_(True),
|
|
|
+ )
|
|
|
+ .order_by(topic_build_item_source.c.id)
|
|
|
)
|
|
|
- .order_by(topic_build_item_source.c.id)
|
|
|
- )
|
|
|
+ ).scalars()
|
|
|
)
|
|
|
- .mappings()
|
|
|
- .all()
|
|
|
if item_ids
|
|
|
else []
|
|
|
)
|
|
|
+ source_rows = await _rows_by_primary_key(
|
|
|
+ self._sessions, topic_build_item_source, [int(value) for value in source_ids]
|
|
|
+ )
|
|
|
|
|
|
if any(
|
|
|
int(row["point_id"]) not in point_ids or int(row["item_id"]) not in item_ids
|
|
|
@@ -241,11 +359,8 @@ class LegacySqlAlchemyInputReader:
|
|
|
"build_id",
|
|
|
"point_type",
|
|
|
"point_result",
|
|
|
- "reason",
|
|
|
"is_active",
|
|
|
"note",
|
|
|
- "created_at",
|
|
|
- "updated_at",
|
|
|
)
|
|
|
item_fields = (
|
|
|
"id",
|
|
|
@@ -259,7 +374,6 @@ class LegacySqlAlchemyInputReader:
|
|
|
"category_id",
|
|
|
"derivation_type",
|
|
|
"step",
|
|
|
- "reason",
|
|
|
"sort_order",
|
|
|
"is_active",
|
|
|
"note",
|
|
|
@@ -280,6 +394,8 @@ class LegacySqlAlchemyInputReader:
|
|
|
"created_at",
|
|
|
)
|
|
|
build_dict = _row_dict(build, build_fields)
|
|
|
+ if not self._include_raw_upload:
|
|
|
+ build_dict["raw_upload_data"] = None
|
|
|
build_dict["demand_constraints"] = _mapping(build_dict["demand_constraints"])
|
|
|
build_dict["agent_config"] = _mapping(build_dict["agent_config"])
|
|
|
build_dict["strategies_config"] = _mapping(build_dict["strategies_config"])
|