feat: AI Worker 平台 MVP v1.0.0 - 多租户/项目管理/AI Worker/任务编排/HITL审核/成本治理
This commit is contained in:
@@ -0,0 +1,29 @@
|
||||
"""Tenant context: ensures all queries are tenant-scoped."""
|
||||
from typing import Optional
|
||||
from fastapi import Request
|
||||
|
||||
|
||||
class TenantContext:
|
||||
"""Thread-local-ish tenant context (stored on request state)."""
|
||||
|
||||
@staticmethod
|
||||
def get_tenant_id(request: Request) -> Optional[int]:
|
||||
"""Extract tenant_id from request state (set by middleware)."""
|
||||
return getattr(request.state, "tenant_id", None)
|
||||
|
||||
@staticmethod
|
||||
def get_user_id(request: Request) -> Optional[int]:
|
||||
"""Extract user_id from request state."""
|
||||
return getattr(request.state, "user_id", None)
|
||||
|
||||
@staticmethod
|
||||
def get_user_role(request: Request) -> Optional[str]:
|
||||
"""Extract user role from request state."""
|
||||
return getattr(request.state, "user_role", None)
|
||||
|
||||
|
||||
def apply_tenant_filter(query, model, tenant_id: int):
|
||||
"""Apply tenant_id filter to a SQLAlchemy query if the model has tenant_id."""
|
||||
if hasattr(model, "tenant_id"):
|
||||
return query.filter(model.tenant_id == tenant_id)
|
||||
return query
|
||||
Reference in New Issue
Block a user