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
/liveand/readyendpoints 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 -
/echois 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
HEALTHCHECKinstruction - 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)