51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
from dataclasses import dataclass
|
|
from uuid import UUID
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.auth.keycloak_dependencies import get_current_keycloak_sub
|
|
from app.core.config import settings
|
|
from app.infra.db.metadata.database import get_metadata_session
|
|
from app.infra.repositories.metadata_repository import MetadataRepository
|
|
|
|
|
|
async def get_metadata_repository(
|
|
session: AsyncSession = Depends(get_metadata_session),
|
|
) -> MetadataRepository:
|
|
return MetadataRepository(session)
|
|
|
|
|
|
async def get_current_metadata_user(
|
|
keycloak_sub: UUID = Depends(get_current_keycloak_sub),
|
|
metadata_repo: MetadataRepository = Depends(get_metadata_repository),
|
|
):
|
|
if settings.AUTH_DISABLED:
|
|
return _AuthBypassUser()
|
|
user = await metadata_repo.get_user_by_keycloak_id(keycloak_sub)
|
|
if not user or not user.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user"
|
|
)
|
|
return user
|
|
|
|
|
|
async def get_current_metadata_admin(
|
|
user=Depends(get_current_metadata_user),
|
|
):
|
|
if settings.AUTH_DISABLED:
|
|
return user
|
|
if user.is_superuser or user.role == "admin":
|
|
return user
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _AuthBypassUser:
|
|
id: UUID = UUID(int=0)
|
|
role: str = "admin"
|
|
is_superuser: bool = True
|
|
is_active: bool = True
|