50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
from fastapi import Depends, Request, params
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from common.aspect.pre_auth import CurrentUserDependency
|
|
from common.context import RequestContext
|
|
from config.get_db import get_db
|
|
from exceptions.exception import ServiceWarning
|
|
from module_admin.entity.vo.user_vo import CurrentUserModel
|
|
from module_business.domain.test_batch_scope import TestBatchScope
|
|
from module_business.service.test_batch_access_service import TestBatchAccessService
|
|
|
|
|
|
def parse_optional_positive_batch_id(raw: str | None) -> int | None:
|
|
"""解析可选的测试批次请求头。"""
|
|
if raw is None:
|
|
return None
|
|
if not isinstance(raw, str) or not raw or not raw.isascii() or not raw.isdigit():
|
|
raise ServiceWarning('', '测试批次标识必须为正整数')
|
|
try:
|
|
batch_id = int(raw)
|
|
except ValueError:
|
|
raise ServiceWarning('', '测试批次标识必须为正整数') from None
|
|
if batch_id <= 0:
|
|
raise ServiceWarning('', '测试批次标识必须为正整数')
|
|
return batch_id
|
|
|
|
|
|
class TestBatchScopeResolver:
|
|
"""将请求头解析为已验证的测试批次作用域。"""
|
|
|
|
def __init__(self, *, write: bool = False) -> None:
|
|
self.write = write
|
|
|
|
async def __call__(
|
|
self,
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: CurrentUserModel = CurrentUserDependency(),
|
|
) -> TestBatchScope:
|
|
raw = request.headers.get('X-Test-Batch-Id')
|
|
batch_id = parse_optional_positive_batch_id(raw)
|
|
scope = await TestBatchAccessService.resolve_scope(db, current_user, batch_id, write=self.write)
|
|
RequestContext.set_current_test_batch_scope(scope)
|
|
return scope
|
|
|
|
|
|
def TestBatchScopeDependency(*, write: bool = False) -> params.Depends: # noqa: N802
|
|
"""构造可配置读写语义的批次作用域依赖。"""
|
|
return Depends(TestBatchScopeResolver(write=write))
|