f0c4060aed
- Common models (Stock, DailyPrice, Valuation, ScreeningResult) - Database connectors (PostgreSQL via asyncpg, MongoDB via motor, Redis) - Redis Streams pub/sub utilities - Base collector class with common patterns - Logging configuration with structlog Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
99 lines
2.8 KiB
Python
99 lines
2.8 KiB
Python
"""Shared FastAPI app factory and common endpoints for all microservices.
|
|
|
|
Each service calls `create_app()` to get a FastAPI instance with:
|
|
- /health - liveness/readiness check
|
|
- /streams - Redis stream queue lengths
|
|
"""
|
|
|
|
import time
|
|
from contextlib import asynccontextmanager
|
|
from typing import AsyncGenerator, Callable, Awaitable
|
|
|
|
import structlog
|
|
import uvicorn
|
|
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
from stock_common.queue.redis_client import get_redis, close_redis
|
|
|
|
logger = structlog.get_logger(module="api_base")
|
|
|
|
_start_time: float = 0.0
|
|
|
|
|
|
def create_app(
|
|
title: str,
|
|
version: str = "0.1.0",
|
|
streams: list[str] | None = None,
|
|
on_startup: Callable[[], Awaitable[None]] | None = None,
|
|
on_shutdown: Callable[[], Awaitable[None]] | None = None,
|
|
) -> FastAPI:
|
|
"""Create a FastAPI app with common health/stream endpoints.
|
|
|
|
Args:
|
|
title: Service name (e.g., "stock-dart-collector").
|
|
version: API version.
|
|
streams: Redis stream names this service reads/writes.
|
|
on_startup: Optional async callback run on startup.
|
|
on_shutdown: Optional async callback run on shutdown.
|
|
"""
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
|
|
global _start_time
|
|
_start_time = time.time()
|
|
if on_startup:
|
|
await on_startup()
|
|
logger.info("api_started", service=title)
|
|
yield
|
|
if on_shutdown:
|
|
await on_shutdown()
|
|
await close_redis()
|
|
logger.info("api_stopped", service=title)
|
|
|
|
app = FastAPI(title=title, version=version, lifespan=lifespan)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
uptime = time.time() - _start_time
|
|
try:
|
|
r = await get_redis()
|
|
await r.ping()
|
|
redis_ok = True
|
|
except Exception:
|
|
redis_ok = False
|
|
return {
|
|
"service": title,
|
|
"status": "ok" if redis_ok else "degraded",
|
|
"redis": redis_ok,
|
|
"uptime_seconds": round(uptime, 1),
|
|
}
|
|
|
|
@app.get("/streams")
|
|
async def stream_info():
|
|
if not streams:
|
|
return {"streams": {}}
|
|
r = await get_redis()
|
|
info = {}
|
|
for stream_name in streams:
|
|
try:
|
|
length = await r.xlen(stream_name)
|
|
info[stream_name] = {"length": length}
|
|
except Exception:
|
|
info[stream_name] = {"length": -1, "error": "stream not found"}
|
|
return {"streams": info}
|
|
|
|
return app
|
|
|
|
|
|
def run_api(app: FastAPI, host: str = "0.0.0.0", port: int = 8000) -> None:
|
|
"""Run the FastAPI app with uvicorn (blocking)."""
|
|
uvicorn.run(app, host=host, port=port, log_level="info")
|