Xiaoxia AI
61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
"""
|
|
Authentication dependency compatibility layer.
|
|
|
|
Canonical bearer-token parsing lives in app.auth. This module remains only so
|
|
legacy imports have a safe target while workspace dependencies are rebuilt.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from app.auth import AuthenticatedUser
|
|
from app.auth import get_current_user as get_authenticated_user
|
|
from app.dependencies import get_user_repository
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from packages.domain.entities import User
|
|
from packages.ports.user_repository import UserRepository
|
|
|
|
optional_bearer_scheme = HTTPBearer(auto_error=False)
|
|
|
|
|
|
async def get_current_user(
|
|
authenticated_user: AuthenticatedUser = Depends(get_authenticated_user),
|
|
) -> User:
|
|
return authenticated_user.user
|
|
|
|
|
|
async def get_current_user_optional(
|
|
credentials: HTTPAuthorizationCredentials | None = Depends(optional_bearer_scheme),
|
|
user_repository: UserRepository = Depends(get_user_repository),
|
|
) -> User | None:
|
|
if credentials is None:
|
|
return None
|
|
try:
|
|
authenticated_user = await get_authenticated_user(credentials, user_repository)
|
|
except HTTPException:
|
|
return None
|
|
return authenticated_user.user
|
|
|
|
|
|
def require_workspace_access(workspace_id: str, user: User = Depends(get_current_user)) -> tuple[str, str]:
|
|
raise _workspace_dependency_not_ready(workspace_id, user.id)
|
|
|
|
|
|
def require_workspace_admin(workspace_id: str, user: User = Depends(get_current_user)) -> str:
|
|
raise _workspace_dependency_not_ready(workspace_id, user.id)
|
|
|
|
|
|
def require_workspace_owner(workspace_id: str, user: User = Depends(get_current_user)) -> str:
|
|
raise _workspace_dependency_not_ready(workspace_id, user.id)
|
|
|
|
|
|
def _workspace_dependency_not_ready(workspace_id: str, user_id: str) -> HTTPException:
|
|
return HTTPException(
|
|
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
|
detail=(
|
|
"Workspace authorization dependencies require SQLAlchemy workspace-member "
|
|
f"repository wiring before use: workspace_id={workspace_id}, user_id={user_id}"
|
|
),
|
|
)
|