feat: health check

Author: antonko

README.md

@@ -15,6 +15,7 @@
 - Поддержка `chat/completions` API с потоковой передачей сообщений
 - Поддержка файлов в сообщениях исключая повторные загрузки в GigaChat
 - Docker-образ
+- Healthcheck API (ready, live)
 
 ### Запуск
 

src/core/settings.py

@@ -1,7 +1,18 @@
+import tomllib
+
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
 
+def get_version() -> str:
+    """Получает версию из pyproject.toml"""
+    try:
+        with open("pyproject.toml", "rb") as f:
+            return tomllib.load(f)["project"]["version"]
+    except (FileNotFoundError, KeyError):
+        return "unknown"
+
+
 class AppSettings(BaseSettings):
     debug: bool = False
     environment: str = "production"
@@ -10,6 +21,7 @@ class AppSettings(BaseSettings):
         description="Bearer token is required to authorize requests.",
     )
     cors_allowed_hosts: list[str] | None = ["http://localhost:5173"]
+    version: str = Field(default_factory=get_version)
 
     class Config:
         env_file = ".env"

src/endpoints.py

@@ -3,10 +3,12 @@
 from fastapi.security import HTTPBearer
 
 from .core.logging import local_logger
+from .core.settings import AppSettings, get_app_settings
 from .core.verify_token import verify_token
 from .gigachat_service import gigachat_service
 from .models.completion import ChatCompletionRequest, ChatCompletionResponse
 from .models.files import FilePurpose, FileUploadResponse
+from .models.health import HealthResponse
 from .models.models import ListModelsResponse
 
 router = APIRouter()
@@ -63,3 +65,25 @@ async def upload_file(
         purpose=purpose.value if purpose != FilePurpose.FINE_TUNE else "general",
     )
     return uploaded
+
+
+@router.get("/health/liveness", response_model=HealthResponse)
+async def liveness(settings: AppSettings = Depends(get_app_settings)) -> HealthResponse:
+    """
+    Liveness probe для kubernetes.
+    Проверяет что сервис жив и отвечает на запросы.
+    """
+    return HealthResponse(status="ok", version=settings.version)
+
+
+@router.get("/health/readiness", response_model=HealthResponse)
+async def readiness(
+    settings: AppSettings = Depends(get_app_settings),
+) -> HealthResponse:
+    """
+    Readiness probe для kubernetes.
+    Проверяет что сервис готов обрабатывать запросы,
+    включая проверку подключения к GigaChat API.
+    """
+    await gigachat_service.get_models()
+    return HealthResponse(status="ok", version=settings.version)

src/endpoints/health.py

@@ -0,0 +1 @@
+ 

src/main.py

@@ -1,3 +1,4 @@
+import gigachat.exceptions
 from fastapi import FastAPI
 from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
@@ -18,6 +19,7 @@ def get_application() -> FastAPI:
         description="Connector to GigaChat API using OpenAI",
         debug=settings.debug,
         cors_allowed_origins=settings.cors_allowed_hosts,
+        version=settings.version,
     )
 
     if settings.cors_allowed_hosts:
@@ -70,6 +72,19 @@ async def exception_handler(request, exc):
             ).model_dump(),
         )
 
+    @app.exception_handler(gigachat.exceptions.ResponseError)
+    async def response_error_handler(request, exc):
+        return JSONResponse(
+            status_code=500,
+            content=ErrorResponse(
+                error=ErrorDetail(
+                    message=str(exc),
+                    type="http",
+                    code="HTTP_EXCEPTION",
+                )
+            ).model_dump(),
+        )
+
     return app
 
 

src/models/health.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel, Field
+
+
+class HealthResponse(BaseModel):
+    status: str = Field("ok", description="The status of the service")
+    version: str = Field(..., description="The version of the service")

tests/conftest.py

@@ -9,6 +9,7 @@
 TEST_BEARER_TOKEN = "test_token"
 os.environ["BEARER_TOKEN"] = TEST_BEARER_TOKEN
 os.environ["GIGACHAT_CREDENTIALS"] = "test_credentials"
+os.environ["DEBUG"] = "false"
 
 
 from src.main import get_application  # noqa: E402

tests/test_health.py

@@ -0,0 +1,51 @@
+from fastapi import status
+from pytest_httpx import HTTPXMock
+
+
+def test_liveness(client):
+    """Test liveness endpoint returns 200 and correct version"""
+    response = client.get("/health/liveness")
+    assert response.status_code == status.HTTP_200_OK
+    data = response.json()
+    assert data["status"] == "ok"
+    assert isinstance(data["version"], str)
+
+
+def test_readiness(client, httpx_mock: HTTPXMock):
+    """Test readiness endpoint returns 200 when GigaChat API is available"""
+    # Mock GigaChat models API call
+    httpx_mock.add_response(
+        json={
+            "data": [
+                {
+                    "id": "GigaChat",
+                    "object": "model",
+                    "owned_by": "salutedevices",
+                    "created": 1735689600,
+                }
+            ],
+            "object": "list",
+        },
+        url="https://gigachat.devices.sberbank.ru/api/v1/models",
+    )
+
+    response = client.get("/health/readiness")
+    assert response.status_code == status.HTTP_200_OK
+    data = response.json()
+    assert data["status"] == "ok"
+    assert isinstance(data["version"], str)
+
+
+def test_readiness_failure(client, httpx_mock: HTTPXMock):
+    """Test readiness endpoint returns 500 when GigaChat API is unavailable"""
+    # Mock GigaChat models API call failure
+    httpx_mock.add_response(
+        status_code=500,
+        url="https://gigachat.devices.sberbank.ru/api/v1/models",
+    )
+
+    response = client.get("/health/readiness")
+    assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+    data = response.json()
+    assert data["error"]["type"] == "http"
+    assert data["error"]["code"] == "HTTP_EXCEPTION"