47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
"""Shared helper functions for signal source APIs."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Optional
|
||
|
||
from fastapi import HTTPException
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ...models import Project
|
||
|
||
|
||
async def resolve_target_project_id(
|
||
*,
|
||
db: AsyncSession,
|
||
tenant_id: int,
|
||
project_id: Optional[int],
|
||
) -> int:
|
||
"""解析目标项目：优先显式 project_id，其次只在唯一 active 项目时自动绑定。"""
|
||
if project_id is not None:
|
||
result = await db.execute(
|
||
select(Project.id).where(
|
||
Project.tenant_id == tenant_id,
|
||
Project.id == int(project_id),
|
||
).limit(1)
|
||
)
|
||
found = result.scalar_one_or_none()
|
||
if found is None:
|
||
raise HTTPException(status_code=404, detail=f"project_id={project_id} 不存在或无权限")
|
||
return int(found)
|
||
|
||
active_result = await db.execute(
|
||
select(Project.id).where(
|
||
Project.tenant_id == tenant_id,
|
||
Project.status == "active",
|
||
).limit(2)
|
||
)
|
||
active_ids = list(active_result.scalars().all())
|
||
if len(active_ids) == 1:
|
||
return int(active_ids[0])
|
||
|
||
raise HTTPException(
|
||
status_code=422,
|
||
detail="未找到唯一目标项目，请在请求中显式传入 project_id",
|
||
)
|