In modern application development, health checks play a crucial role in ensuring reliability, observability, and smooth orchestration—especially in containerized environments like Docker or Kubernetes. In this post, I’ll walk you through how I built a production-ready health-check microservice using FastAPI.
This project features structured logging, clean separation of concerns, and asynchronous service checks for both a database and Redis—all built in a modular and extensible way.
GitHub Repo: [https://github.com/DanielPopoola/fastapi-microservice-health-check]
🚀 What This Project Covers
- Creating a
/health/
endpoint with real service checks (DB, Redis) - Supporting
/live
and/ready
endpoints for Kubernetes probes - Using async
asyncio.gather()
for fast, parallel checks - Configurable settings with Pydantic
- Structured logging with custom log formatting using loguru.
- Middleware for request timing and error handling
📁 Project Structure
project/
├── main.py # App factory and configuration
├── config.py # App settings via Pydantic
├── routers/
│ ├── health.py # Health check endpoints
│ └── echo.py # Echo endpoint (for demo)
├── utils/
│ └── logging.py # Custom logger setup
└── ...
🔍 Under the Hood: main.py
main.py
acts as the orchestrator. Here's what it handles:
1. App Lifecycle Management
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("Application starting up")
yield
logger.info("Application shutting down")
This cleanly logs startup and shutdown events, essential for container lifecycle awareness.
2. App Factory Pattern
The create_app()
function encapsulates app setup:
- Loads settings with
get_settings()
- Sets up structured logging
- Registers CORS middleware
- Adds global and HTTP exception handlers
- Includes routers for modularity
3. Middleware
A custom middleware logs request data and execution time:
@app.middleware("http")
async def log_requests(request, call_next):
start_time = time.time()
response = await call_next(request)
response.headers["X-Response-Time"] = f"{(time.time() - start_time) * 1000:.2f}ms"
return response
4. Exception Handling
Two global handlers catch errors and format them consistently:
- One for
HTTPException
- One for unexpected
Exception
⚕️ Health Check Logic (routers/health.py
)
The routers/health.py
file houses the core of this service:
✅ /health/
Performs parallel health checks using asyncio.gather()
:
async def perform_health_checks(settings: Settings) -> Dict[str, ServiceCheck]:
checks = {}
tasks = []
if settings.database_url:
tasks.append(("database", check_database(settings.database_url, settings.health_check_timeout)))
if settings.redis_url:
tasks.append(("redis", check_redis(settings.redis_url, settings.health_check_timeout)))
if tasks:
results = await asyncio.gather(*[task[1] for task in tasks], return_exceptions=True)
...
return checks
The result is a combined status response showing the health of each component.
🔁 /live
A simple liveness check returning HTTP 200 to signal the app is alive.
📦 /ready
Waits for both Redis and DB to pass checks before returning 200. Useful for Kubernetes readiness probes.
📡 Root Endpoint and Echo
-
/
returns app metadata like name, version, and timestamp -
/echo
is a simple test endpoint to verify connectivity
🛠️ How to Run It
uvicorn app.main:app --reload
Or using the embedded __main__
block:
python -m main
🌟 What’s Next?
- Add more service checks (e.g., external APIs, caches)
- Integrate with Docker’s
HEALTHCHECK
instruction - Configure Kubernetes readiness/liveness probes
🧠 Final Thoughts
Building robust health checks is one of the simplest yet most impactful ways to improve system reliability. With FastAPI’s speed and async support, this project offers a solid base for both simple and enterprise-grade applications.
Top comments (0)