feat(api): enhance authentication and health check endpoints with detailed responses and descriptions

This commit is contained in:
Abhimanyu Saharan
2026-02-15 02:35:31 +05:30
parent 7e147ee608
commit ae17facf88
4 changed files with 155 additions and 21 deletions

View File

@@ -5,13 +5,56 @@ from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, status
from app.core.auth import AuthContext, get_auth_context
from app.schemas.errors import LLMErrorResponse
from app.schemas.users import UserRead
router = APIRouter(prefix="/auth", tags=["auth"])
AUTH_CONTEXT_DEP = Depends(get_auth_context)
@router.post("/bootstrap", response_model=UserRead)
@router.post(
"/bootstrap",
response_model=UserRead,
summary="Bootstrap Authenticated User Context",
description=(
"Resolve caller identity from auth headers and return the canonical user profile. "
"This endpoint does not accept a request body."
),
responses={
status.HTTP_200_OK: {
"description": "Authenticated user profile resolved from token claims.",
"content": {
"application/json": {
"example": {
"id": "11111111-1111-1111-1111-111111111111",
"clerk_user_id": "user_2abcXYZ",
"email": "alex@example.com",
"name": "Alex Chen",
"preferred_name": "Alex",
"pronouns": "they/them",
"timezone": "America/Los_Angeles",
"notes": "Primary operator for board triage.",
"context": "Handles incident coordination and escalation.",
"is_super_admin": False,
}
}
},
},
status.HTTP_401_UNAUTHORIZED: {
"model": LLMErrorResponse,
"description": "Caller is not authenticated as a user actor.",
"content": {
"application/json": {
"example": {
"detail": {"code": "unauthorized", "message": "Not authenticated"},
"code": "unauthorized",
"retryable": False,
}
}
},
},
},
)
async def bootstrap_user(auth: AuthContext = AUTH_CONTEXT_DEP) -> UserRead:
"""Return the authenticated user profile from token claims."""
if auth.actor_type != "user" or auth.user is None:

View File

@@ -5,7 +5,7 @@ from __future__ import annotations
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING
from fastapi import APIRouter, FastAPI
from fastapi import APIRouter, FastAPI, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi_pagination import add_pagination
@@ -34,6 +34,7 @@ from app.core.config import settings
from app.core.error_handling import install_error_handling
from app.core.logging import configure_logging, get_logger
from app.db.session import init_db
from app.schemas.health import HealthStatusResponse
if TYPE_CHECKING:
from collections.abc import AsyncIterator
@@ -124,22 +125,58 @@ else:
install_error_handling(app)
@app.get("/health", tags=["health"])
def health() -> dict[str, bool]:
@app.get(
"/health",
tags=["health"],
response_model=HealthStatusResponse,
summary="Health Check",
description="Lightweight liveness probe endpoint.",
responses={
status.HTTP_200_OK: {
"description": "Service is alive.",
"content": {"application/json": {"example": {"ok": True}}},
}
},
)
def health() -> HealthStatusResponse:
"""Lightweight liveness probe endpoint."""
return {"ok": True}
return HealthStatusResponse(ok=True)
@app.get("/healthz", tags=["health"])
def healthz() -> dict[str, bool]:
@app.get(
"/healthz",
tags=["health"],
response_model=HealthStatusResponse,
summary="Health Alias Check",
description="Alias liveness probe endpoint for platform compatibility.",
responses={
status.HTTP_200_OK: {
"description": "Service is alive.",
"content": {"application/json": {"example": {"ok": True}}},
}
},
)
def healthz() -> HealthStatusResponse:
"""Alias liveness probe endpoint for platform compatibility."""
return {"ok": True}
return HealthStatusResponse(ok=True)
@app.get("/readyz", tags=["health"])
def readyz() -> dict[str, bool]:
@app.get(
"/readyz",
tags=["health"],
response_model=HealthStatusResponse,
summary="Readiness Check",
description="Readiness probe endpoint for service orchestration checks.",
responses={
status.HTTP_200_OK: {
"description": "Service is ready.",
"content": {"application/json": {"example": {"ok": True}}},
}
},
)
def readyz() -> HealthStatusResponse:
"""Readiness probe endpoint for service orchestration checks."""
return {"ok": True}
return HealthStatusResponse(ok=True)
api_v1 = APIRouter(prefix="/api/v1")

View File

@@ -0,0 +1,16 @@
"""Health and readiness probe response schemas."""
from __future__ import annotations
from pydantic import Field
from sqlmodel import SQLModel
class HealthStatusResponse(SQLModel):
"""Standard payload for service liveness/readiness checks."""
ok: bool = Field(
description="Indicates whether the probe check succeeded.",
examples=[True],
)

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
from uuid import UUID
from pydantic import Field
from sqlmodel import SQLModel
RUNTIME_ANNOTATION_TYPES = (UUID,)
@@ -12,14 +13,45 @@ RUNTIME_ANNOTATION_TYPES = (UUID,)
class UserBase(SQLModel):
"""Common user profile fields shared across user payload schemas."""
clerk_user_id: str
email: str | None = None
name: str | None = None
preferred_name: str | None = None
pronouns: str | None = None
timezone: str | None = None
notes: str | None = None
context: str | None = None
clerk_user_id: str = Field(
description="External auth provider user identifier (Clerk).",
examples=["user_2abcXYZ"],
)
email: str | None = Field(
default=None,
description="Primary email address for the user.",
examples=["alex@example.com"],
)
name: str | None = Field(
default=None,
description="Full display name.",
examples=["Alex Chen"],
)
preferred_name: str | None = Field(
default=None,
description="Preferred short name used in UI.",
examples=["Alex"],
)
pronouns: str | None = Field(
default=None,
description="Preferred pronouns.",
examples=["they/them"],
)
timezone: str | None = Field(
default=None,
description="IANA timezone identifier.",
examples=["America/Los_Angeles"],
)
notes: str | None = Field(
default=None,
description="Internal notes for operators.",
examples=["Primary operator for board triage."],
)
context: str | None = Field(
default=None,
description="Additional context used by the system for personalization.",
examples=["Handles incident coordination and escalation."],
)
class UserCreate(UserBase):
@@ -40,5 +72,11 @@ class UserUpdate(SQLModel):
class UserRead(UserBase):
"""Full user payload returned by API responses."""
id: UUID
is_super_admin: bool
id: UUID = Field(
description="Internal user UUID.",
examples=["11111111-1111-1111-1111-111111111111"],
)
is_super_admin: bool = Field(
description="Whether this user has tenant-wide super-admin privileges.",
examples=[False],
)