Source code for app.domain.system.controllers

"""System domain controllers."""

from __future__ import annotations

from typing import TYPE_CHECKING, Literal, TypeVar

import structlog
from litestar import Controller, MediaType, get
from litestar.response import Response
from sqlalchemy import text

from app.domain.system import schemas as s

if TYPE_CHECKING:
    from sqlalchemy.ext.asyncio import AsyncSession

    from app.lib.settings import AppSettings

logger = structlog.get_logger()
OnlineOffline = TypeVar("OnlineOffline", bound=Literal["online", "offline"])


[docs] class SystemController(Controller): """System health and configuration.""" tags = ["System"] dependencies = {} @get( operation_id="SystemHealth", name="system:health", path="/health", summary="Health Check", exclude_from_auth=True, security=[], # Public endpoint - no auth required ) async def check_system_health( self, db_session: AsyncSession, settings: AppSettings, ) -> Response[s.SystemHealth]: """Check database available and returns app config info. Args: db_session: The database session. settings: Application settings. Returns: The response object. """ db_status: Literal["online", "offline"] try: await db_session.execute(text("select 1")) db_status = "online" except ConnectionRefusedError: db_status = "offline" healthy = db_status == "online" if healthy: await logger.adebug( "System Health", database_status=db_status, ) else: await logger.awarn( "System Health Check", database_status=db_status, ) return Response( content=s.SystemHealth(database_status=db_status, app=settings.NAME), status_code=200 if healthy else 500, media_type=MediaType.JSON, ) @get( operation_id="OAuthConfig", name="system:oauth-config", path="/api/config/oauth", summary="Get OAuth Configuration", exclude_from_auth=True, security=[], # Public endpoint - no auth required ) async def get_oauth_config(self, settings: AppSettings) -> s.OAuthConfig: """Get OAuth provider configuration for frontend. Args: settings: Application settings. Returns: OAuth configuration indicating which providers are enabled. """ return s.OAuthConfig(google_enabled=settings.google_oauth_enabled, github_enabled=settings.github_oauth_enabled)