agent-platform/services/auth-service/app/api/routes.py

from typing import Annotated

from core_domain import ServiceHealth
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.application.services import AuthApplicationService
from app.db.session import get_db
from app.domain.repositories import RoleAssignmentRepository, RoleRepository, UserRepository
from app.schemas.auth import (
    LoginRequest,
    LoginResponse,
    PermissionCheckRequest,
    PermissionCheckResponse,
    RoleAssignmentCreateRequest,
    RoleAssignmentResponse,
    RoleAssignmentStatusUpdateRequest,
    RoleCreateRequest,
    RoleResponse,
    RoleStatusUpdateRequest,
    TokenVerifyRequest,
    TokenVerifyResponse,
    UserCreateRequest,
    UserResponse,
    UserStatusUpdateRequest,
)

router = APIRouter()
DbSession = Annotated[Session, Depends(get_db)]
UserIdQuery = Annotated[str, Query(...)]


def get_auth_application_service(request: Request, db: DbSession) -> AuthApplicationService:
    settings = request.app.state.settings
    return AuthApplicationService(
        user_repository=UserRepository(db),
        role_repository=RoleRepository(db),
        assignment_repository=RoleAssignmentRepository(db),
        token_secret=settings.credential_encryption_key)


AuthServiceDep = Annotated[AuthApplicationService, Depends(get_auth_application_service)]


@router.get("/health", response_model=ServiceHealth)
def health_check(db: DbSession) -> ServiceHealth:
    db.execute(text("SELECT 1"))
    return ServiceHealth(service="auth-service", status="ok", database="ok")


@router.post("/login", response_model=LoginResponse)
def login(
    payload: LoginRequest,
    service: AuthServiceDep) -> LoginResponse:
    result = service.login(payload)
    if result is None:
        raise HTTPException(status_code=401, detail="invalid username or password")
    return result


@router.post("/tokens/verify", response_model=TokenVerifyResponse)
def verify_token(
    payload: TokenVerifyRequest,
    service: AuthServiceDep) -> TokenVerifyResponse:
    return service.verify_token(payload)


@router.post("/users", response_model=UserResponse)
def create_user(
    payload: UserCreateRequest,
    service: AuthServiceDep) -> UserResponse:
    return UserResponse.from_entity(service.create_user(payload))


@router.get("/users", response_model=list[UserResponse])
def list_users(
    service: AuthServiceDep) -> list[UserResponse]:
    return [UserResponse.from_entity(item) for item in service.list_users()]


@router.patch("/users/{user_id}/status", response_model=UserResponse)
def update_user_status(
    user_id: str,
    payload: UserStatusUpdateRequest,
    service: AuthServiceDep) -> UserResponse:
    entity = service.update_user_status(user_id=user_id, payload=payload)
    if entity is None:
        raise HTTPException(status_code=404, detail=f"user not found: {user_id}")
    return UserResponse.from_entity(entity)


@router.post("/roles", response_model=RoleResponse)
def create_role(
    payload: RoleCreateRequest,
    service: AuthServiceDep) -> RoleResponse:
    return RoleResponse.from_entity(service.create_role(payload))


@router.get("/roles", response_model=list[RoleResponse])
def list_roles(
    service: AuthServiceDep) -> list[RoleResponse]:
    return [RoleResponse.from_entity(item) for item in service.list_roles()]


@router.patch("/roles/{role_id}/status", response_model=RoleResponse)
def update_role_status(
    role_id: str,
    payload: RoleStatusUpdateRequest,
    service: AuthServiceDep) -> RoleResponse:
    entity = service.update_role_status(role_id=role_id, payload=payload)
    if entity is None:
        raise HTTPException(status_code=404, detail=f"role not found: {role_id}")
    return RoleResponse.from_entity(entity)


@router.post("/assignments", response_model=RoleAssignmentResponse)
def create_assignment(
    payload: RoleAssignmentCreateRequest,
    service: AuthServiceDep) -> RoleAssignmentResponse:
    return RoleAssignmentResponse.from_entity(service.create_assignment(payload))


@router.get("/assignments", response_model=list[RoleAssignmentResponse])
def list_assignments(
    user_id: UserIdQuery,
    service: AuthServiceDep) -> list[RoleAssignmentResponse]:
    return [
        RoleAssignmentResponse.from_entity(item)
        for item in service.list_assignments(user_id=user_id)
    ]


@router.patch("/assignments/{assignment_id}/status", response_model=RoleAssignmentResponse)
def update_assignment_status(
    assignment_id: str,
    payload: RoleAssignmentStatusUpdateRequest,
    service: AuthServiceDep) -> RoleAssignmentResponse:
    entity = service.update_assignment_status(assignment_id=assignment_id, payload=payload)
    if entity is None:
        raise HTTPException(status_code=404, detail=f"assignment not found: {assignment_id}")
    return RoleAssignmentResponse.from_entity(entity)


@router.post("/permissions/check", response_model=PermissionCheckResponse)
def check_permission(
    payload: PermissionCheckRequest,
    service: AuthServiceDep) -> PermissionCheckResponse:
    return service.check_permission(payload)