Spaces:

Shwy1001
/

BioPatch

Sleeping

Jerry

Hardware bridge: ESP32 device ingest + hardware mode

34cb656 3 months ago

3.33 kB

from collections.abc import Generator

from sqlalchemy import create_engine, event
from sqlalchemy.engine import Engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from sqlalchemy.pool import StaticPool

from core.config import settings


_is_sqlite = settings.DATABASE_URL.startswith("sqlite")
_is_memory = _is_sqlite and (":memory:" in settings.DATABASE_URL or "mode=memory" in settings.DATABASE_URL)
_connect_args = {"check_same_thread": False} if _is_sqlite else {}


def configure_sqlite(bound_engine: Engine, *, in_memory: bool | None = None) -> None:
    """Apply the SQLite pragmas the concurrent ingest path relies on.
    Alert-time Jarvis enrichment now commits from a worker thread while the
    ingest coroutine (and other live requests) hold their own connections, so
    two writers can race for the database lock. Without a busy timeout SQLite
    fails that race immediately with "database is locked"; with one, the loser
    waits its (brief) turn instead. WAL additionally keeps a write from
    blocking dashboard reads — but it is only valid on an on-disk database, so
    we skip it for in-memory test databases.
    """
    if in_memory is None:
        url = str(bound_engine.url)
        in_memory = ":memory:" in url or "mode=memory" in url

    @event.listens_for(bound_engine, "connect")
    def _set_sqlite_pragmas(dbapi_conn, _record):  # pragma: no cover - trivial
        cursor = dbapi_conn.cursor()
        cursor.execute("PRAGMA busy_timeout=5000")
        if not in_memory:
            cursor.execute("PRAGMA journal_mode=WAL")
            cursor.execute("PRAGMA synchronous=NORMAL")
        cursor.close()


_engine_kwargs: dict = {
    "connect_args": _connect_args,
    "echo": settings.DEBUG,
    "future": True,
}
if _is_memory:
    # A shared in-memory SQLite database exists only while a connection to it
    # is open. Now that alert-time enrichment opens its own session from a
    # worker thread, a QueuePool could momentarily hold zero connections and
    # destroy the database mid-run. StaticPool keeps a single long-lived
    # connection so the schema survives across threads (SQLite serializes
    # access internally). On-disk deployments keep the default pool.
    _engine_kwargs["poolclass"] = StaticPool

engine = create_engine(settings.DATABASE_URL, **_engine_kwargs)

if _is_sqlite:
    configure_sqlite(engine, in_memory=_is_memory)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=Session)


class Base(DeclarativeBase):
    pass


def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


def init_db() -> None:
    # Importing here ensures every model is registered against Base.metadata
    # before create_all walks the registry.
    from models import Base as ModelsBase  # noqa: F401

    Base.metadata.create_all(bind=engine)

    # Auto-add columns introduced after the demo DB was first created.
    # Safe + idempotent — no-op when the schema is already up to date.
    try:
        from core.migrations import upgrade_schema

        upgrade_schema(engine)
    except Exception:  # pragma: no cover - never block startup
        import logging

        logging.getLogger("biopatch").exception("Auto-migration failed")

	from collections.abc import Generator

	from sqlalchemy import create_engine, event
	from sqlalchemy.engine import Engine
	from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
	from sqlalchemy.pool import StaticPool

	from core.config import settings


	_is_sqlite = settings.DATABASE_URL.startswith("sqlite")
	_is_memory = _is_sqlite and (":memory:" in settings.DATABASE_URL or "mode=memory" in settings.DATABASE_URL)
	_connect_args = {"check_same_thread": False} if _is_sqlite else {}


	def configure_sqlite(bound_engine: Engine, *, in_memory: bool \| None = None) -> None:
	"""Apply the SQLite pragmas the concurrent ingest path relies on.

	Alert-time Jarvis enrichment now commits from a worker thread while the
	ingest coroutine (and other live requests) hold their own connections, so
	two writers can race for the database lock. Without a busy timeout SQLite
	fails that race immediately with "database is locked"; with one, the loser
	waits its (brief) turn instead. WAL additionally keeps a write from
	blocking dashboard reads — but it is only valid on an on-disk database, so
	we skip it for in-memory test databases.
	"""
	if in_memory is None:
	url = str(bound_engine.url)
	in_memory = ":memory:" in url or "mode=memory" in url

	@event.listens_for(bound_engine, "connect")
	def _set_sqlite_pragmas(dbapi_conn, _record): # pragma: no cover - trivial
	cursor = dbapi_conn.cursor()
	cursor.execute("PRAGMA busy_timeout=5000")
	if not in_memory:
	cursor.execute("PRAGMA journal_mode=WAL")
	cursor.execute("PRAGMA synchronous=NORMAL")
	cursor.close()


	_engine_kwargs: dict = {
	"connect_args": _connect_args,
	"echo": settings.DEBUG,
	"future": True,
	}
	if _is_memory:
	# A shared in-memory SQLite database exists only while a connection to it
	# is open. Now that alert-time enrichment opens its own session from a
	# worker thread, a QueuePool could momentarily hold zero connections and
	# destroy the database mid-run. StaticPool keeps a single long-lived
	# connection so the schema survives across threads (SQLite serializes
	# access internally). On-disk deployments keep the default pool.
	_engine_kwargs["poolclass"] = StaticPool

	engine = create_engine(settings.DATABASE_URL, **_engine_kwargs)

	if _is_sqlite:
	configure_sqlite(engine, in_memory=_is_memory)

	SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=Session)


	class Base(DeclarativeBase):
	pass


	def get_db() -> Generator[Session, None, None]:
	db = SessionLocal()
	try:
	yield db
	finally:
	db.close()


	def init_db() -> None:
	# Importing here ensures every model is registered against Base.metadata
	# before create_all walks the registry.
	from models import Base as ModelsBase # noqa: F401

	Base.metadata.create_all(bind=engine)

	# Auto-add columns introduced after the demo DB was first created.
	# Safe + idempotent — no-op when the schema is already up to date.
	try:
	from core.migrations import upgrade_schema

	upgrade_schema(engine)
	except Exception: # pragma: no cover - never block startup
	import logging

	logging.getLogger("biopatch").exception("Auto-migration failed")