Spaces:
Running
Running
File size: 9,182 Bytes
79b0bef 4dc0836 79b0bef | 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 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 | """
src/db/connection.py
ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
Database connection and session management.
This module is the only place in the codebase that knows which
database engine is in use. Everything above it (ETL, API,
repositories) calls get_session() and works with the result β
they never build connection strings or reference a dialect.
How the backend is selected
ββββββββββββββββββββββββββββ
The DATABASE_URL environment variable drives the choice:
Not set / .env missing
β SQLite file at project root (clinical_nlp.db)
β Zero configuration; works out of the box
DATABASE_URL=sqlite:///./clinical_nlp.db
β Same SQLite file, explicit
DATABASE_URL=postgresql://user:pass@host:5432/dbname
β PostgreSQL (Supabase, AWS RDS, local Postgres, anything)
One variable. No code changes.
SQLite vs PostgreSQL quirks
ββββββββββββββββββββββββββββ
SQLAlchemy handles most dialect differences transparently, but
two things need special handling:
1. Connection pool: SQLite is file-based and single-writer;
the NullPool prevents "database is locked" errors when
multiple threads try to connect simultaneously.
2. check_same_thread=False: Required for SQLite when used
with FastAPI (which runs handlers in a thread pool).
Both are applied automatically based on the URL.
ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
"""
from __future__ import annotations
from collections.abc import Generator
from contextlib import contextmanager
from sqlalchemy import create_engine, event, text
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import NullPool, StaticPool
from src.utils.config import DatabaseConfig
from src.utils.logger import get_logger
logger = get_logger(__name__)
def _build_engine(url: str) -> Engine:
"""Create a SQLAlchemy engine appropriate for the given URL.
Applies connection pool settings that work correctly for both
SQLite (development) and PostgreSQL (staging / production).
Args:
url: SQLAlchemy-compatible database URL.
Returns:
Configured :class:`sqlalchemy.engine.Engine` instance.
"""
is_sqlite = url.startswith("sqlite")
is_sqlite_memory = is_sqlite and ":memory:" in url
if is_sqlite_memory:
# In-memory SQLite is connection-private: each new DBAPI
# connection gets its own fresh, empty database. NullPool opens
# a new connection on every checkout, so tables created via
# create_all_tables() on one connection would be invisible to
# the next (e.g. the connection handling an API request) --
# this caused "no such table" errors in the test suite.
# StaticPool keeps exactly one connection alive for the engine's
# lifetime so every checkout shares the same in-memory database.
engine = create_engine(
url,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
logger.debug("SQLite in-memory engine created: %s", url)
elif is_sqlite:
# NullPool avoids the "database is locked" error that occurs
# when SQLite is accessed from multiple threads (e.g. FastAPI).
# Safe for a file-based DB since the file on disk persists
# state across connections (unlike :memory:, see above).
engine = create_engine(
url,
connect_args={"check_same_thread": False},
poolclass=NullPool,
)
# Enable WAL mode for better concurrent read performance
@event.listens_for(engine, "connect")
def set_wal_mode(dbapi_conn, _):
dbapi_conn.execute("PRAGMA journal_mode=WAL")
logger.debug("SQLite engine created: %s", url)
else:
# PostgreSQL β use connection pooling for efficiency
engine = create_engine(
url,
pool_size = DatabaseConfig.pool_size,
max_overflow = DatabaseConfig.max_overflow,
pool_timeout = DatabaseConfig.pool_timeout,
# Recycle connections after 30 minutes to avoid
# "server closed connection" errors on long-running apps
pool_recycle = 1800,
)
logger.debug("PostgreSQL engine created")
return engine
# ββ Module-level singletons βββββββββββββββββββββββββββββββββββββββ
# Created once at import time. Tests can call _reset() to swap
# in an in-memory SQLite database without restarting the process.
_engine: Engine | None = None
_SessionFactory: sessionmaker | None = None
def get_engine() -> Engine:
"""Return the module-level database engine, creating it if needed.
Returns:
The active :class:`~sqlalchemy.engine.Engine`.
"""
global _engine
if _engine is None:
_engine = _build_engine(DatabaseConfig.url)
return _engine
def get_session_factory() -> sessionmaker:
"""Return the module-level session factory, creating it if needed.
Returns:
A :class:`~sqlalchemy.orm.sessionmaker` bound to the engine.
"""
global _SessionFactory
if _SessionFactory is None:
_SessionFactory = sessionmaker(
bind = get_engine(),
autocommit = False,
autoflush = False,
expire_on_commit = False, # safer for async contexts
)
return _SessionFactory
@contextmanager
def get_session() -> Generator[Session, None, None]:
"""Provide a transactional database session as a context manager.
Commits on clean exit; rolls back and re-raises on any exception.
Always closes the session when the block exits.
Yields:
An active :class:`~sqlalchemy.orm.Session`.
Example::
with get_session() as session:
note = session.get(ClinicalNote, note_id)
note.severity = "urgent"
# committed automatically on clean exit
# Exception example:
with get_session() as session:
session.add(bad_record)
# β rolls back; exception propagates to caller
"""
factory = get_session_factory()
session = factory()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
def get_db_session() -> Generator[Session, None, None]:
"""FastAPI dependency that yields a database session per request.
Designed for use with FastAPI's ``Depends()``. Closes the
session after the response is sent, even on errors.
Yields:
An active :class:`~sqlalchemy.orm.Session`.
Example::
@router.get("/notes/{note_id}")
def read_note(note_id: int, db: Session = Depends(get_db_session)):
return db.get(ClinicalNote, note_id)
"""
factory = get_session_factory()
session = factory()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
def create_all_tables() -> None:
"""Create all database tables defined in the ORM models.
Safe to call multiple times β uses ``checkfirst=True`` so
existing tables are not dropped or modified.
Typically called once at application startup.
"""
from src.db.models import Base # imported here to avoid circular imports
Base.metadata.create_all(bind=get_engine(), checkfirst=True)
logger.info("Database tables created (or already exist)")
def check_connection() -> bool:
"""Verify that the database is reachable and responding.
Returns:
True if the connection succeeds; False otherwise.
Example::
if not check_connection():
raise RuntimeError("Database unreachable at startup")
"""
try:
with get_engine().connect() as conn:
conn.execute(text("SELECT 1"))
logger.info("Database connection verified β")
return True
except Exception as exc:
logger.error("Database connection failed: %s", exc)
return False
def _reset_for_testing(url: str = "sqlite:///:memory:") -> None:
"""Replace the engine with a fresh in-memory database.
Only intended for use in the test suite. Do not call in
production code.
Args:
url: Database URL for the test engine.
Defaults to an in-memory SQLite database.
"""
global _engine, _SessionFactory
if _engine:
_engine.dispose()
_engine = _build_engine(url)
_SessionFactory = sessionmaker(
bind=_engine, autocommit=False, autoflush=False
)
logger.debug("Test database engine reset: %s", url)
|