Engine

Handles SQLModel engine creation and session management for CodeMap.

logger module-attribute

logger = getLogger(__name__)

get_engine async

get_engine(echo: bool = False) -> Engine

Gets or creates the SQLAlchemy engine for the database.

Tries to use PostgreSQL first, falling back to SQLite in-memory if PostgreSQL is not available. Ensures the PostgreSQL Docker container is running before creating the engine.

Parameters:

Name	Type	Description	Default
echo	bool	Whether to echo SQL statements to the log.	False

Returns:

Name	Type	Description
Engine	Engine	The SQLAlchemy Engine instance.

Raises:

Type	Description
RuntimeError	If neither PostgreSQL nor SQLite can be initialized.

Source code in src/codemap/db/engine.py

async def get_engine(echo: bool = False) -> Engine:
	"""
	Gets or creates the SQLAlchemy engine for the database.

	Tries to use PostgreSQL first, falling back to SQLite in-memory if PostgreSQL is not available.
	Ensures the PostgreSQL Docker container is running before creating the engine.

	Args:
	        echo (bool): Whether to echo SQL statements to the log.

	Returns:
	        Engine: The SQLAlchemy Engine instance.

	Raises:
	        RuntimeError: If neither PostgreSQL nor SQLite can be initialized.

	"""
	# Check if we should skip Docker and use SQLite (for testing or non-Docker environments)
	use_sqlite = os.environ.get("CODEMAP_USE_SQLITE", "").lower() in ("true", "1", "yes")

	if use_sqlite:
		return _create_sqlite_engine(echo)

	# Try PostgreSQL first
	try:
		return await _create_postgres_engine(echo)
	except (RuntimeError, ConnectionError, TimeoutError, OSError) as e:
		logger.warning(f"PostgreSQL connection failed: {e}. Falling back to SQLite in-memory database.")
		return _create_sqlite_engine(echo)

create_db_and_tables

create_db_and_tables(engine_instance: Engine) -> None

Creates the database and all tables defined in SQLModel models.

Source code in src/codemap/db/engine.py

def create_db_and_tables(engine_instance: Engine) -> None:
	"""Creates the database and all tables defined in SQLModel models."""
	logger.info("Ensuring database tables exist...")
	try:
		SQLModel.metadata.create_all(engine_instance)
		logger.info("Database tables ensured.")
	except Exception:
		logger.exception("Error creating database tables")
		raise

get_session

get_session(
	engine_instance: Engine,
) -> Generator[Session, None, None]

Provides a context-managed SQLModel session from a given engine.

Source code in src/codemap/db/engine.py

@contextmanager
def get_session(engine_instance: Engine) -> Generator[Session, None, None]:
	"""Provides a context-managed SQLModel session from a given engine."""
	session = Session(engine_instance)
	try:
		yield session
	except Exception:
		logger.exception("Session rollback due to exception")
		session.rollback()
		raise
	finally:
		session.close()