Improve health schemas and readiness checks

This commit is contained in:
2025-11-27 14:21:25 +01:00
parent 7acce1da02
commit 3a7208d28d
4 changed files with 246 additions and 100 deletions

View File

@@ -1,23 +1,65 @@
"""Pydantic schemas for health responses."""
"""
Pydantic schemas for defining health check responses, following IETF standards.
"""
from datetime import datetime
from enum import Enum
from pydantic import BaseModel
# Import ConfigDict from pydantic to resolve the deprecation warning
from pydantic import BaseModel, Field, ConfigDict
# Define acceptable statuses as an Enum for robust validation
class HealthStatusEnum(str, Enum):
"""Enumeration for standard health check statuses."""
passed = "pass"
warned = "warn"
failed = "fail"
class ComponentCheck(BaseModel):
name: str
# pass | warn | fail
status: str
time: datetime | None = None
output: str | None = None
observedValue: float | int | None = None
observedUnit: str | None = None
"""
Represents the status and metrics for a single internal component or dependency.
"""
# Use ConfigDict instead of the class Config approach
model_config = ConfigDict(
populate_by_name=True # Allows instantiation using either 'observed_value' or 'observedValue'
)
name: str = Field(description="The unique name of the component being checked (e.g., 'postgres', 'redis').")
status: HealthStatusEnum = Field(description="The status of the check: 'pass', 'warn', or 'fail'.")
time: datetime | None = Field(default=None, description="The time at which the check was performed in ISO 8601 format.")
output: str | None = Field(default=None, description="Additional details, error messages, or logs if the status is 'fail' or 'warn'.")
# Python uses snake_case internally, JSON uses camelCase for the alias
observed_value: float | int | None = Field(
default=None,
alias="observedValue",
description="The value observed during the check (e.g., latency in ms)."
)
# Python uses snake_case internally, JSON uses camelCase for the alias
observed_unit: str | None = Field(
default=None,
alias="observedUnit",
description="The unit of the observed value (e.g., 'ms', 'count', 'bytes')."
)
class HealthStatus(BaseModel):
# pass | warn | fail
status: str
version: str | None = None
environment: str | None = None
serviceName: str | None = None
description: str | None = None
checks: dict[str, ComponentCheck]
"""
The overall system health response model, aggregating all individual component checks.
"""
# Use ConfigDict instead of the class Config approach
model_config = ConfigDict(
populate_by_name=True
)
status: HealthStatusEnum = Field(description="The aggregate status of the entire service: 'pass', 'warn', or 'fail'.")
version: str | None = Field(default=None, description="The application version (e.g., Git SHA or semantic version number).")
environment: str | None = Field(default=None, description="The deployment environment (e.g., 'production', 'staging').")
# Python uses snake_case internally, JSON uses camelCase for the alias
service_name: str | None = Field(
default=None,
alias="serviceName",
description="The name of the service."
)
description: str | None = Field(default=None, description="A brief description of the service.")
checks: dict[str, ComponentCheck] = Field(description="A dictionary mapping check keys (e.g., 'Database') to their detailed ComponentCheck results.")