安全:增加测试批次上下文硬校验
This commit is contained in:
@@ -0,0 +1,33 @@
|
||||
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}不属于当前数据模式或测试批次')
|
||||
Reference in New Issue
Block a user