185 lines
6.8 KiB
Python
185 lines
6.8 KiB
Python
"""持久化任务创建和重试，不在 HTTP 事务中读取源库。"""
|
||
|
||
import secrets
|
||
from contextlib import nullcontext
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
|
||
from sqlalchemy.exc import IntegrityError
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import get_settings
|
||
from app.core.exceptions import BadRequestError, NotFoundError
|
||
from app.core.tplus_db import resolve_tplus_settings
|
||
from app.models import TPlusSyncBatch, TPlusSyncError, User
|
||
from app.repositories import tplus_repository as repository
|
||
from app.schemas.tplus import TPlusCreateIn, TPlusRetryIn
|
||
from app.services import audit_service
|
||
from app.services.audit_service import AuditRecord, RequestMeta
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TaskActor:
|
||
"""HTTP 操作者与请求元信息，CLI 调用不传。"""
|
||
|
||
user: User
|
||
meta: RequestMeta | None = None
|
||
|
||
|
||
def new_batch_no() -> str:
|
||
"""生成最长 24 字符的可追溯批次号。"""
|
||
return f"TP{datetime.now():%Y%m%d%H%M%S}{secrets.token_hex(4)}"
|
||
|
||
|
||
def _audit(session, actor, action, data):
|
||
if actor is None:
|
||
return nullcontext()
|
||
return audit_service.audit(
|
||
session,
|
||
operator=actor.user,
|
||
meta=actor.meta,
|
||
record=AuditRecord("T+数据同步", action, "T+ 数据任务" + action, data),
|
||
)
|
||
|
||
|
||
def _source_database() -> str:
|
||
try:
|
||
return resolve_tplus_settings(get_settings()).t_plus_database
|
||
except RuntimeError as exc:
|
||
raise BadRequestError(str(exc)) from None
|
||
|
||
|
||
def _same_intent(batch, mode, entities, parent):
|
||
if (
|
||
batch.sync_mode != mode
|
||
or sorted(batch.entity_types_json or []) != sorted(entities)
|
||
or batch.parent_batch_no != parent
|
||
):
|
||
raise BadRequestError("幂等键已用于不同的任务请求")
|
||
return batch
|
||
|
||
|
||
def _insert(session, values):
|
||
try:
|
||
with session.begin_nested():
|
||
batch = TPlusSyncBatch(**values)
|
||
session.add(batch)
|
||
session.flush()
|
||
return batch, True
|
||
except IntegrityError:
|
||
batch = repository.get_request(session, values["request_key"], for_update=True)
|
||
if batch is None:
|
||
raise
|
||
return _same_intent(
|
||
batch, values["sync_mode"], values["entity_types_json"], values["parent_batch_no"]
|
||
), False
|
||
|
||
|
||
def _values(payload, database, actor, parent=None):
|
||
return {
|
||
"batch_no": new_batch_no(),
|
||
"request_key": str(payload.request_key),
|
||
"status": "QUEUED",
|
||
"sync_mode": "AUDIT" if payload.task_type == "AUDIT" else "FULL",
|
||
"entity_types_json": sorted(payload.entity_types),
|
||
"source_database": database,
|
||
"requested_by": actor.user.id if actor else None,
|
||
"parent_batch_no": parent,
|
||
"started_at": None,
|
||
"source_record_count": 0,
|
||
"staged_record_count": 0,
|
||
"failed_record_count": 0,
|
||
"skipped_record_count": 0,
|
||
}
|
||
|
||
|
||
def create_task(session: Session, payload: TPlusCreateIn, actor: TaskActor | None = None):
|
||
"""原子创建任务；重放已存在的意图不依赖此刻的源连接状态。"""
|
||
with _audit(session, actor, "创建读取任务", payload.model_dump(mode="json")):
|
||
existing = repository.get_request(session, str(payload.request_key))
|
||
mode = "AUDIT" if payload.task_type == "AUDIT" else "FULL"
|
||
if existing:
|
||
batch = _same_intent(existing, mode, payload.entity_types, None)
|
||
else:
|
||
batch, _ = _insert(session, _values(payload, _source_database(), actor))
|
||
session.commit()
|
||
return batch
|
||
|
||
|
||
def retry_task(
|
||
session: Session, parent_no: str, payload: TPlusRetryIn, actor: TaskActor | None = None
|
||
):
|
||
"""锁父批次后创建一个新任务；不覆盖原批次及异常历史。"""
|
||
with _audit(
|
||
session,
|
||
actor,
|
||
"重试读取任务",
|
||
{"parent_batch_no": parent_no, "request_key": str(payload.request_key)},
|
||
):
|
||
existing = repository.get_request(session, str(payload.request_key))
|
||
if existing:
|
||
if existing.parent_batch_no != parent_no:
|
||
raise BadRequestError("幂等键已用于不同的重试请求")
|
||
session.commit()
|
||
batch = existing
|
||
else:
|
||
parent = repository.get_batch(session, parent_no, for_update=True)
|
||
if parent is None:
|
||
raise NotFoundError("同步批次不存在")
|
||
request = _retry_request(parent, payload)
|
||
active = repository.active_retry(session, parent_no)
|
||
if active:
|
||
if active.request_key != str(payload.request_key):
|
||
raise BadRequestError("该批次已有未完成的重试任务，请刷新列表查看")
|
||
batch = active
|
||
else:
|
||
batch, inserted = _insert(
|
||
session, _values(request, _source_database(), actor, parent_no)
|
||
)
|
||
if inserted:
|
||
repository.increment_retry_errors(session, parent_no)
|
||
session.commit()
|
||
return batch
|
||
|
||
|
||
def _retry_request(parent, payload):
|
||
if parent.status not in ("FAILED", "PARTIAL"):
|
||
raise BadRequestError("只能重试失败或部分失败的批次")
|
||
if parent.sync_mode not in ("FULL", "AUDIT"):
|
||
raise BadRequestError("该同步方式尚未启用，请重新选择读取任务")
|
||
if parent.sync_mode != "AUDIT" and parent.entity_types_json is None:
|
||
raise BadRequestError("历史批次未记录范围，请重新选择读取类型")
|
||
return TPlusCreateIn(
|
||
task_type="AUDIT" if parent.sync_mode == "AUDIT" else "READ",
|
||
entity_types=parent.entity_types_json or [],
|
||
request_key=payload.request_key,
|
||
)
|
||
|
||
|
||
def fail_batch(session: Session, batch: TPlusSyncBatch, code: str) -> None:
|
||
"""只保存固定错误摘要；即使原异常含连接密码也不直接落库。"""
|
||
messages = {
|
||
"INTERRUPTED": "执行进程中断，请重新读取；已暂存数据会自动去重",
|
||
"EXECUTION_ERROR": "任务执行失败，请检查只读连接和字段映射后重试",
|
||
"TYPE_UNRESOLVED": "核验所需单据类型缺失或不唯一，请核对源字典",
|
||
}
|
||
message = messages.get(code, "任务未完成，请核对运行配置")
|
||
batch.status, batch.finished_at, batch.error_message = "FAILED", datetime.now(), message
|
||
batch.failed_record_count = (batch.failed_record_count or 0) + 1
|
||
session.add(
|
||
TPlusSyncError(
|
||
batch_no=batch.batch_no,
|
||
entity_type="system",
|
||
source_table="system",
|
||
error_code=code,
|
||
error_message=message,
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
|
||
def recover_running(session: Session) -> None:
|
||
"""仅在持有执行锁时调用；锁证明之前的执行者已经退出。"""
|
||
for batch in repository.running_batches(session):
|
||
fail_batch(session, batch, "INTERRUPTED")
|