Akiba So
fc468464b2
Context: Build a spatial risk assessment system correlating air quality data with children's respiratory disease incidence across Wuhan. Approach: FastAPI backend serving PostGIS spatial queries, React frontend with Deck.gl maps, and a PyTorch SpatialTemporalGCN pipeline for multi-day (1d/3d/7d) risk prediction. Changes: - backend/ — FastAPI API with auth (JWT), alerts, risk analysis, geocoded case data, grid statistics, and report endpoints - frontend/ — React dashboard with interactive risk maps, alert monitoring, district comparison charts, and timeline player - models/ — SpatialTemporalGCN model with trained weights and ONNX export for inference - scripts/ — ETL pipeline for weather + medical data, grid generation, feature engineering, training, and daily inference - deploy/ — Docker Compose configs for backend, frontend, and MLflow - docs/ — API docs, deployment guide, user guide, and code review Impact: Enables spatial risk visualization, alert monitoring, and ML-driven health risk forecasting for environmental health teams.
95 lines
2.8 KiB
Python
95 lines
2.8 KiB
Python
"""
|
|
Database connection and session management for PostGIS
|
|
"""
|
|
import logging
|
|
import asyncpg
|
|
from typing import Optional
|
|
from contextlib import asynccontextmanager
|
|
from pydantic_settings import BaseSettings
|
|
|
|
logger = logging.getLogger("cbpoa.database")
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
"""Database settings from environment variables"""
|
|
POSTGRES_HOST: str = "localhost"
|
|
POSTGRES_PORT: int = 5432
|
|
POSTGRES_USER: str = ""
|
|
POSTGRES_PASSWORD: str = ""
|
|
POSTGRES_DB: str = ""
|
|
|
|
class Config:
|
|
env_file = ".env"
|
|
|
|
|
|
settings = Settings()
|
|
|
|
if not settings.POSTGRES_USER or not settings.POSTGRES_PASSWORD or not settings.POSTGRES_DB:
|
|
raise RuntimeError(
|
|
"Missing required database environment variables: POSTGRES_USER, POSTGRES_PASSWORD, POSTGRES_DB. "
|
|
"Create a .env file or set them in your environment."
|
|
)
|
|
|
|
|
|
class Database:
|
|
"""Async database connection pool manager"""
|
|
|
|
def __init__(self):
|
|
self.pool: Optional[asyncpg.Pool] = None
|
|
|
|
async def connect(self):
|
|
"""Initialize database connection pool"""
|
|
if self.pool is None:
|
|
dsn = f"postgresql://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}"
|
|
self.pool = await asyncpg.create_pool(
|
|
dsn=dsn,
|
|
min_size=5,
|
|
max_size=20,
|
|
command_timeout=60
|
|
)
|
|
logger.info("Database connection pool created successfully")
|
|
|
|
async def disconnect(self):
|
|
"""Close database connection pool"""
|
|
if self.pool:
|
|
await self.pool.close()
|
|
self.pool = None
|
|
logger.info("Database connection pool closed")
|
|
|
|
@asynccontextmanager
|
|
async def get_connection(self):
|
|
"""Get a connection from the pool"""
|
|
if self.pool is None:
|
|
await self.connect()
|
|
|
|
async with self.pool.acquire() as connection:
|
|
yield connection
|
|
|
|
@asynccontextmanager
|
|
async def get_transaction(self):
|
|
"""Get a transaction context"""
|
|
if self.pool is None:
|
|
await self.connect()
|
|
|
|
async with self.pool.acquire() as connection:
|
|
async with connection.transaction():
|
|
yield connection
|
|
|
|
|
|
# Global database instance
|
|
db = Database()
|
|
|
|
|
|
async def init_db():
|
|
"""Initialize database on startup - graceful degradation if unavailable"""
|
|
try:
|
|
await db.connect()
|
|
except Exception as e:
|
|
logger.warning("Database not available (%s). Running in demo mode.", e)
|
|
logger.warning("Set POSTGRES_HOST/POSTGRES_USER/POSTGRES_PASSWORD environment variables for database access.")
|
|
|
|
|
|
async def close_db():
|
|
"""Close database on shutdown"""
|
|
await db.disconnect()
|