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)