Initial commit

.github/workflows/workflow.yml

@@ -0,0 +1,18 @@
+name: CI Pipeline
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+  release:
+    types:
+      - published
+
+
+jobs:
+  ci:
+    uses: community-of-python/community-workflow/.github/workflows/preset.yml@main
+    secrets: inherit

.gitignore

@@ -0,0 +1,66 @@
+.idea/
+.vscode/
+# Byte-compiled / optimized / DLL files
+pycache/
+__pycache__
+*.py[cod]
+*$py.class
+__test*.py
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+tmp/
+__test.py
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Environments
+.env
+.gen.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Cython debug symbols
+cython_debug/

README.md

@@ -0,0 +1,175 @@
+# HealthChecks
+
+Welcome to the healthiest library of all times! It provides a simple interface to check the health of your application.
+
+We have base classes for HTTP and FILE based health checks.
+
+# Installation
+
+TODO
+
+### If you want to check health of your **FastAPI** application, run:
+
+```bash
+poetry run health-checks -E fastapi
+```
+
+### If you want to check health of your **Litestar** application, run:
+
+```bash
+poetry run health-checks -E litestar
+```
+
+### If you want to check health of your **consumer**, run:
+
+```bash
+poetry run health-checks -E file
+```
+
+## HTTP based quickstart
+
+Let's begin with http based healthchecks for **Litestar** application:
+
+```python
+from health_checks.http_based import DefaultHTTPHealthCheck
+from health_checks.litestar_healthcheck import build_litestar_health_check_router
+import litestar
+
+
+litestar_application = litestar.Litestar(
+    route_handlers=[
+        build_litestar_health_check_router(
+            healthcheck_endpoint="/health/",
+            health_check=DefaultHTTPHealthCheck(),
+        ),
+    ],
+)
+```
+
+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.
+
+Similar to litestar, here is the **FastAPI** example
+
+```python
+import fastapi
+from health_checks.fastapi_healthcheck import build_fastapi_health_check_router
+from health_checks.http_based import DefaultHTTPHealthCheck
+
+
+fastapi_app = fastapi.FastAPI()
+fastapi_app.include_router(
+    build_fastapi_health_check_router(
+        health_check_endpoint="/health/",
+        health_check=DefaultHTTPHealthCheck(),
+    ),
+)
+```
+
+This is also it! How wonderful, isn't it? You can navigate to `/health/` and meet your 200 HTTP status code.
+
+## FILE based quickstart
+
+Here things are starting to get complicated.
+Let's imagine a simple consumer
+
+```python
+import dataclasses
+
+from health_cheks.base import HealthCheck
+
+
+@dataclasses.dataclass
+class SimpleConsumer:
+    health_check: HealthCheck
+
+    async def startup(self):
+        await self.health_check.startup()
+
+    async def shutdown(self):
+        await self.health_check.shutdown()
+
+    async def listen(self):
+        while True:
+            # Here we receive our messages from some queue
+            try:
+                # Non-blocking message processing
+                await self.process_message()
+
+                # Be attentive! We call update_health method, not update_health_status.
+                await health_check.update_health()
+            except Exception:
+                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.  
+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
+
+```python
+# directory/some_file.py
+import asyncio
+
+from health_checks import file_based
+
+
+health_check_object = file_based.DefaultFileHealthCheck()
+consumer = SimpleConsumer(health_check_object)
+
+if __name__ == '__main__':
+    asyncio.run(consumer.run_comsumer())
+```
+
+Cool! Now during your consumer process health will be updated. But how to check it and where?
+
+In this package we have a cli, that allows you to check health of certain **HealthCheck** object. Here, how you can use it
+
+```bash
+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
+
+And you use it inside your k8s manifest like this:
+
+```yaml
+livenessProbe:
+  exec:
+    command:
+      - python
+      - "-m"
+      - health_checks
+      - directory.some_file:health_check_object
+```
+
+Now let's look at FILE health check accepted arguments.
+
+```python
+@dataclasses.dataclass
+class BaseFileHealthCheck(base.HealthCheck):
+    failure_threshold: int = 60
+    health_check_period: int = 30
+    healthcheck_file_name: str | None = None
+    base_folder: str = "./tmp/health-checks"
+    ...
+```
+
+- `base_folder` - folder, where health check file will be created.
+- `failure_threshold` - time after which health check won't pass
+- `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.  
+> 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?**  
+  This parameter helps to throttle calls to `update_health` method. By default `update_health` will be called every 30 seconds.
+- **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

@@ -0,0 +1,42 @@
+from __future__ import annotations
+import os
+import typing
+
+
+POD_IDENTIFIER_ENVIRONMENT_NAME: typing.Final = "HOSTNAME"
+HEALTH_CHECK_FILE_NAME_TEMPLATE: typing.Final = "health-check-{file_name}.json"
+
+
+class HealthCheckTypedDict(typing.TypedDict, total=False):
+    service_version: typing.Optional[str]
+    service_name: typing.Optional[str]
+    health_status: bool
+
+
+class HealthCheck(typing.Protocol):
+    service_version_env: str
+    service_name_env: str
+    service_version: typing.Optional[str]
+    service_name: typing.Optional[str]
+
+    def _get_health_check_data(
+        self,
+        health_status: bool,
+    ) -> HealthCheckTypedDict:
+        return {
+            "service_version": self.service_version or os.environ.get(self.service_version_env),
+            "service_name": self.service_name or os.environ.get(self.service_name_env),
+            "health_status": health_status,
+        }
+
+    async def update_health_status(self) -> bool:
+        raise NotImplementedError
+
+    async def check_health(self) -> HealthCheckTypedDict:
+        raise NotImplementedError
+
+    async def startup(self) -> None:
+        pass
+
+    async def shutdown(self) -> None:
+        pass

health_checks/fastapi_healthcheck.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+import typing
+
+from fastapi import APIRouter
+from fastapi.exceptions import HTTPException
+from fastapi.responses import JSONResponse
+
+from health_checks.base import HealthCheck
+
+
+def build_fastapi_health_check_router(
+    health_check: HealthCheck,
+    health_check_endpoint: str = "/health/",
+) -> APIRouter:
+    fastapi_router: typing.Final = APIRouter(tags=["probes"])
+
+    @fastapi_router.get(health_check_endpoint)
+    async def health_check_handler() -> JSONResponse:
+        health_check_data: typing.Final = await health_check.check_health()
+        if not health_check_data["health_status"]:
+            raise HTTPException(status_code=500, detail="Service is unhealthy.")
+        return JSONResponse(content=health_check_data)
+
+    return fastapi_router