company-ai-platform/app/modules/legacy_mysql/services/task_sync.py

from typing import Any

from fastapi import HTTPException, status
from sqlalchemy import select

from app.core.constants import ActorValue
from app.core.utils.time import utc_now
from app.modules.audit.constants import AuditAction, AuditRiskLevel, AuditSource, AuditStatus
from app.modules.audit.schemas import AuditLogCreate
from app.modules.audit.service import AuditService
from app.modules.business.constants import BusinessDomain, SourceSystem, StatusValue
from app.modules.business.models import LegacySyncRun, WorkTask
from app.modules.business.service import serialize_model
from app.modules.events.constants import (
    EventAggregateType,
    EventPayloadKey,
    EventSource,
    EventType,
)
from app.modules.events.services import EventService
from app.modules.legacy_mysql.constants import (
    LEGACY_SYNC_MISSING_ID_REASON,
    LEGACY_TASK_QUERY_SOURCE,
    LEGACY_TASK_SYNC_NOTE,
    LEGACY_TASK_SYNC_RUN_CODE_PREFIX,
    LegacyQueryError,
    LegacyQueryName,
    LegacyResponseKey,
    LegacySyncAction,
    LegacyTaskField,
)

from app.modules.legacy_mysql.services.common import _query_name_text


class LegacyTaskSyncMixin:
    def sync_tasks(
        self,
        source_query: str | None = None,
        source_query_name: str | None = None,
        field_map: dict[str, str] | None = None,
        limit: int = 100,
        dry_run: bool = True,
        actor: str = ActorValue.API,
    ) -> dict[str, Any]:
        if self.db is None:
            raise HTTPException(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                detail=LegacyQueryError.APP_DB_UNAVAILABLE,
            )

        query_name = source_query_name or LegacyQueryName.TASKS
        if source_query:
            rows = self.execute_readonly(
                source_query,
                {LegacyResponseKey.LIMIT: limit},
                limit=limit,
            )[LegacyResponseKey.ROWS]
            query_ref = LegacySyncAction.ALLOWLISTED_INLINE_SQL
        else:
            rows = self.execute_allowed_query(
                query_name,
                {LegacyResponseKey.LIMIT: limit},
                limit=limit,
            )[LegacyResponseKey.ROWS]
            query_ref = _query_name_text(query_name)
        field_map = field_map or {}
        created = 0
        updated = 0
        skipped = 0
        items: list[dict[str, Any]] = []

        for row in rows:
            payload = self._task_payload(row, field_map)
            if not payload[LegacyTaskField.EXTERNAL_ID] and not payload[LegacyTaskField.CODE]:
                skipped += 1
                items.append(
                    {
                        LegacyResponseKey.ACTION: LegacySyncAction.SKIPPED,
                        LegacyResponseKey.REASON: LEGACY_SYNC_MISSING_ID_REASON,
                        LegacyResponseKey.SOURCE: row,
                    }
                )
                continue

            stmt = select(WorkTask).where(
                WorkTask.source_system == SourceSystem.LEGACY_MYSQL,
                WorkTask.external_id == payload[LegacyTaskField.EXTERNAL_ID],
            )
            record = self.db.execute(stmt).scalar_one_or_none()
            if record is None:
                record = self.db.execute(
                    select(WorkTask).where(WorkTask.code == payload[LegacyTaskField.CODE])
                ).scalar_one_or_none()

            if record is None:
                created += 1
                action = LegacySyncAction.CREATE
                result = payload
                if not dry_run:
                    record = WorkTask(**payload)
                    self.db.add(record)
                    self.db.flush()
                    result = serialize_model(record)
            else:
                updated += 1
                action = LegacySyncAction.UPDATE
                if not dry_run:
                    for key, value in payload.items():
                        setattr(record, key, value)
                    self.db.flush()
                    result = serialize_model(record)
                else:
                    result = payload
            items.append(
                {
                    LegacyResponseKey.ACTION: action,
                    LegacyResponseKey.TASK: result,
                    LegacyResponseKey.SOURCE: row,
                }
            )

        result = {
            LegacyResponseKey.DRY_RUN: dry_run,
            LegacyResponseKey.CREATED: created,
            LegacyResponseKey.UPDATED: updated,
            LegacyResponseKey.SKIPPED: skipped,
            LegacyResponseKey.ITEMS: items,
        }
        sync_run = LegacySyncRun(
            code=f"{LEGACY_TASK_SYNC_RUN_CODE_PREFIX}-{utc_now():%Y%m%d%H%M%S%f}",
            domain=BusinessDomain.TASKS,
            source_table=LEGACY_TASK_QUERY_SOURCE,
            status=StatusValue.DRY_RUN if dry_run else AuditStatus.SUCCESS,
            finished_at=utc_now(),
            created_count=created,
            updated_count=updated,
            skipped_count=skipped,
            note=LEGACY_TASK_SYNC_NOTE,
        )
        self.db.add(sync_run)
        self.db.flush()
        result[LegacyResponseKey.SYNC_RUN_CODE] = sync_run.code

        AuditService(self.db).record(
            AuditLogCreate(
                actor=actor,
                source=AuditSource.LEGACY_MYSQL,
                action=AuditAction.LEGACY_SYNC_TASKS,
                target_type=BusinessDomain.TASKS,
                risk_level=AuditRiskLevel.MEDIUM,
                request_payload={
                    LegacyResponseKey.SOURCE_QUERY: query_ref,
                    LegacyResponseKey.FIELD_MAP: field_map,
                    LegacyResponseKey.LIMIT: limit,
                    LegacyResponseKey.DRY_RUN: dry_run,
                },
                response_payload={
                    key: result[key]
                    for key in [
                        LegacyResponseKey.DRY_RUN,
                        LegacyResponseKey.CREATED,
                        LegacyResponseKey.UPDATED,
                        LegacyResponseKey.SKIPPED,
                    ]
                },
            )
        )
        EventService(self.db).enqueue(
            event_type=EventType.LEGACY_SYNC_COMPLETED,
            source=EventSource.LEGACY_MYSQL,
            aggregate_type=EventAggregateType.LEGACY_SYNC_RUN,
            aggregate_id=sync_run.code,
            actor=actor,
            payload={
                EventPayloadKey.CODE: sync_run.code,
                EventPayloadKey.DOMAIN: BusinessDomain.TASKS,
                EventPayloadKey.STATUS: sync_run.status,
                EventPayloadKey.CREATED: created,
                EventPayloadKey.UPDATED: updated,
                EventPayloadKey.SKIPPED: skipped,
            },
            idempotency_key=f"legacy-sync:{sync_run.code}",
        )
        self.db.commit()
        self.db.refresh(sync_run)
        return result