Function _get_engine

registry.py:157–184

Get or create the database engine (thread-safe singleton pattern). Returns: Tuple of (engine, SessionLocal)

()

def _get_engine():
    """
    Get or create the database engine (thread-safe singleton pattern).

    Returns:
        Tuple of (engine, SessionLocal)
    """
    global _engine, _SessionLocal

    # Double-checked locking for thread safety
    if _engine is None:
        with _engine_lock:
            if _engine is None:
                db_path = get_registry_path()
                db_url = f"sqlite:///{db_path.as_posix()}"
                _engine = create_engine(
                    db_url,
                    connect_args={
                        "check_same_thread": False,
                        "timeout": SQLITE_TIMEOUT,
                    }
                )
                Base.metadata.create_all(bind=_engine)
                _migrate_add_default_concurrency(_engine)
                _SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=_engine)
                logger.debug("Initialized registry database at: %s", db_path)

    return _engine, _SessionLocal


def _migrate_add_default_concurrency(engine) -> None: