34 lines
1.2 KiB
Python
34 lines
1.2 KiB
Python
from dataclasses import dataclass
|
|
from typing import Any, Literal
|
|
|
|
from sqlalchemy.orm import InstrumentedAttribute
|
|
from sqlalchemy.sql.elements import ColumnElement
|
|
|
|
from exceptions.exception import ServiceException
|
|
from module_business.domain.enums import TestBatchStatus
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TestBatchScope:
|
|
"""当前请求可信任的数据批次作用域。"""
|
|
|
|
batch_id: int | None
|
|
mode: Literal['FORMAL', 'TEST']
|
|
status: TestBatchStatus | None
|
|
writable: bool
|
|
|
|
@classmethod
|
|
def formal(cls) -> 'TestBatchScope':
|
|
return cls(batch_id=None, mode='FORMAL', status=None, writable=True)
|
|
|
|
|
|
def batch_scope_condition(column: InstrumentedAttribute, scope: TestBatchScope) -> ColumnElement[bool]:
|
|
"""生成正式数据或当前测试批次的查询条件。"""
|
|
return column.is_(None) if scope.batch_id is None else column == scope.batch_id
|
|
|
|
|
|
def assert_record_in_scope(record: Any, scope: TestBatchScope, label: str) -> None:
|
|
"""防止已加载记录跨数据模式或测试批次被使用。"""
|
|
if record is None or record.test_batch_id != scope.batch_id:
|
|
raise ServiceException('', f'{label}不属于当前数据模式或测试批次')
|