File size: 2,421 Bytes
1425afc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.exc import OperationalError
# ==========================================================
# DATABASE URL
# ==========================================================
DATABASE_URL = os.getenv("DATABASE_URL")
if not DATABASE_URL:
raise RuntimeError(
"DATABASE_URL environment variable is not set"
)
# ==========================================================
# SUPABASE ENTERPRISE ENGINE CONFIG
# ==========================================================
# Optimized for:
# - Supabase Pooler
# - FastAPI async workload
# - Background workers
# - Long-running autonomous services
engine = create_engine(
DATABASE_URL,
# --- Pool Stability ---
pool_pre_ping=True, # validates dead connections
pool_recycle=300, # refresh connections
pool_size=5, # safe baseline
max_overflow=10, # burst capacity
# --- Reliability ---
echo=False,
future=True,
# --- Supabase Requirement ---
connect_args={
"sslmode": "require",
"connect_timeout": 10,
},
)
# ==========================================================
# SESSION FACTORY
# ==========================================================
SessionLocal = sessionmaker(
autocommit=False,
autoflush=False,
bind=engine,
)
# ==========================================================
# BASE MODEL
# ==========================================================
Base = declarative_base()
# ==========================================================
# DEPENDENCY (FASTAPI)
# ==========================================================
def get_db():
"""
FastAPI dependency injection session.
Ensures connection cleanup even on crash.
"""
db = SessionLocal()
try:
yield db
finally:
db.close()
# ==========================================================
# CONNECTION TEST (STARTUP SAFE)
# ==========================================================
def verify_database_connection():
"""
Validates database connectivity during startup.
Prevents silent runtime failures.
"""
try:
with engine.connect() as conn:
conn.execute("SELECT 1")
except OperationalError as e:
raise RuntimeError(
f"Database connection failed: {str(e)}"
) |