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)