Exclude /health route from OpenAPI schema by default, and allow to configure it

README.md

@@ -46,7 +46,7 @@ litestar_application = litestar.Litestar(
 )
 ```
 
-This is it! Now if yout go to `/health/` you will notice a 200 HTTP status code if everything is allright. Otherwise you will face a 500 HTTP status code.
+This is it! Now if your go to `/health/` you will notice a 200 HTTP status code if everything is alright. Otherwise you will face a 500 HTTP status code.
 
 Similar to litestar, here is the **FastAPI** example
 
@@ -75,7 +75,7 @@ Let's imagine a simple consumer
 ```python
 import dataclasses
 
-from health_cheks.base import HealthCheck
+from health_checks.base import HealthCheck
 
 
 @dataclasses.dataclass
@@ -101,8 +101,8 @@ class SimpleConsumer:
                 continue
 ```
 
-This is very **important** to place your health check inside infinite loop or something like that in your consumer.  
-You cannot use it inside your message processing function or method because if there will be no messages - your consumer will die eventually. And this is not the case we are lookin for.  
+This is very **important** to place your health check inside infinite loop or something like that in your consumer.
+You cannot use it inside your message processing function or method because if there will be no messages - your consumer will die eventually. And this is not the case we are looking for.
 So, your update_health method call should be independent from message processing, also it should not be locked by it.
 
 So, here how your code could look like
@@ -118,7 +118,7 @@ health_check_object = file_based.DefaultFileHealthCheck()
 consumer = SimpleConsumer(health_check_object)
 
 if __name__ == '__main__':
-    asyncio.run(consumer.run_comsumer())
+    asyncio.run(consumer.run_consumer())
 ```
 
 Cool! Now during your consumer process health will be updated. But how to check it and where?
@@ -130,7 +130,7 @@ python -m health_checks directory.some_file:health_check_object
 ```
 
 Here `some_file` is the name of file and `health_check_object` is the name of file_based.DefaultFileHealthCheck object.
-If everything is allright, then there will be no exception, but if it is not - there will be
+If everything is alright, then there will be no exception, but if it is not - there will be
 
 And you use it inside your k8s manifest like this:
 
@@ -161,15 +161,15 @@ class BaseFileHealthCheck(base.HealthCheck):
 - `health_check_period` - delay time before updating health check file
 - `healthcheck_file_name` - you can pass an explicit file name to your health check.
 
-> IMPORTANT: You actually have to pass `healthcheck_file_name` it if your are not running in k8s environment.  
-> In that case your health check file will be named randomly and you cannot check health with provided script.  
+> IMPORTANT: You actually have to pass `healthcheck_file_name` it if your are not running in k8s environment.
+> In that case your health check file will be named randomly and you cannot check health with provided script.
 > If you are running in k8s, then file name will be made of `HOSTNAME` env variable a.k.a. pod id.
 
 > IMPORTANT: Consider putting your health check into separate file to prevent useless imports during health check script execution.
 
 ## FAQ
 
-- **Why do i even need `health_check_period` in FILE based health check?**  
+- **Why do i even need `health_check_period` in FILE based health check?**
   This parameter helps to throttle calls to `update_health` method. By default `update_health` will be called every 30 seconds.
-- **Custom health checks**  
+- **Custom health checks**
   There are two options. You can inherit from `BaseFileHealthCheck` or `BaseHTTPHealthCheck`. Another way is to implement class according to HealthCheck protocol. More information about protocols [here](https://peps.python.org/pep-0544/).

health_checks/base.py

@@ -8,8 +8,8 @@
 
 
 class HealthCheckTypedDict(typing.TypedDict, total=False):
-    service_version: str | None
-    service_name: str | None
+    service_version: typing.Optional[str]  # noqa: UP007 (Litestar fails to build OpenAPI schema on Python 3.9)
+    service_name: typing.Optional[str]  # noqa: UP007 (Litestar fails to build OpenAPI schema on Python 3.9)
     health_status: bool
 
 

health_checks/fastapi_healthcheck.py

@@ -11,10 +11,15 @@
 
 
 def build_fastapi_health_check_router(
+    *,
     health_check: HealthCheck,
     health_check_endpoint: str = "/health/",
+    include_in_schema: bool = False,
 ) -> APIRouter:
-    fastapi_router: typing.Final = APIRouter(tags=["probes"])
+    fastapi_router: typing.Final = APIRouter(
+        tags=["probes"],
+        include_in_schema=include_in_schema,
+    )
 
     @fastapi_router.get(health_check_endpoint)
     async def health_check_handler() -> JSONResponse:

health_checks/litestar_healthcheck.py

@@ -8,8 +8,10 @@
 
 
 def build_litestar_health_check_router(
+    *,
     health_check: base.HealthCheck,
     health_check_endpoint: str = "/health/",
+    include_in_schema: bool = False,
 ) -> litestar.Router:
     @litestar.get(media_type=litestar.MediaType.JSON)
     async def health_check_handler() -> base.HealthCheckTypedDict:
@@ -18,4 +20,9 @@ async def health_check_handler() -> base.HealthCheckTypedDict:
             raise litestar.exceptions.HTTPException(status_code=500, detail="Service is unhealthy.")
         return health_check_data
 
-    return litestar.Router(path=health_check_endpoint, route_handlers=[health_check_handler], tags=["probes"])
+    return litestar.Router(
+        path=health_check_endpoint,
+        route_handlers=[health_check_handler],
+        tags=["probes"],
+        include_in_schema=include_in_schema,
+    )

tests/test_default_health_check.py

@@ -2,12 +2,16 @@
 import asyncio
 import pathlib
 import typing
+from unittest import mock
 from unittest.mock import patch
 
 import fastapi
 import litestar
 import pytest
 
+from health_checks.fastapi_healthcheck import build_fastapi_health_check_router
+from health_checks.litestar_healthcheck import build_litestar_health_check_router
+
 
 if typing.TYPE_CHECKING:
     from httpx import AsyncClient
@@ -63,7 +67,6 @@ async def test_default_file_health_check(
     await short_lived_default_file_health_check.shutdown()
 
 
-@pytest.mark.anyio
 async def test_litestar_healthcheck(
     litestar_client: AsyncClient,
     health_check_endpoint: str,
@@ -82,7 +85,6 @@ async def test_litestar_healthcheck(
         ).status_code == litestar.status_codes.HTTP_500_INTERNAL_SERVER_ERROR
 
 
-@pytest.mark.anyio
 async def test_fastapi_healthcheck(
     fastapi_client: AsyncClient,
     health_check_endpoint: str,
@@ -99,3 +101,24 @@ async def test_fastapi_healthcheck(
         assert (
             await fastapi_client.get(health_check_endpoint)
         ).status_code == fastapi.status.HTTP_500_INTERNAL_SERVER_ERROR
+
+
+@pytest.mark.parametrize("include_in_schema", [True, False])
+def test_litestar_include_in_schema(include_in_schema: bool) -> None:
+    health_check_router: typing.Final = build_litestar_health_check_router(
+        health_check=mock.Mock(), include_in_schema=include_in_schema
+    )
+    application: typing.Final = litestar.Litestar(route_handlers=[health_check_router])
+
+    assert bool(application.openapi_schema.paths) is include_in_schema
+
+
+@pytest.mark.parametrize("include_in_schema", [True, False])
+def test_fastapi_include_in_schema(include_in_schema: bool) -> None:
+    health_check_router = build_fastapi_health_check_router(
+        health_check=mock.Mock(), include_in_schema=include_in_schema
+    )
+    application: typing.Final = fastapi.FastAPI()
+    application.include_router(health_check_router)
+
+    assert bool(application.openapi()["paths"]) is include_in_schema