ahtlon
db87625384
This was anoying to fix (and even debug) so I used the local llm for some code changes Assisted-by: local-qwen3.8 <localllm@ahtlon.de>
91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
import logging
|
|
|
|
from sqlalchemy import inspect
|
|
from sqlmodel import Session, SQLModel, create_engine
|
|
|
|
from app.services.settings import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# the lifespan startup (create_db_and_tables, create_first_user),
|
|
# get_db_session(), the module level scanner session and
|
|
# the get_session() FastAPI dependency resolve their engine
|
|
# through get_engine(). That makes configure_engine() the single
|
|
# injection point for tests
|
|
_engine = None
|
|
_engine_is_default = False
|
|
|
|
|
|
def _build_engine_from_settings():
|
|
return create_engine(
|
|
settings.sqlalchemy_database_url, connect_args={"check_same_thread": False}
|
|
)
|
|
|
|
|
|
def get_engine():
|
|
"""Return the engine, building it from settings on first use."""
|
|
global _engine, _engine_is_default
|
|
if _engine is None:
|
|
_engine = _build_engine_from_settings()
|
|
_engine_is_default = True
|
|
return _engine
|
|
|
|
|
|
def configure_engine(engine) -> None:
|
|
"""Install *engine* as the engine used by this module from now on.
|
|
|
|
The replaced engine is not closed because the scanner (or other long-lived session)
|
|
could still hold connections
|
|
"""
|
|
global _engine, _engine_is_default
|
|
_engine = engine
|
|
_engine_is_default = False
|
|
|
|
|
|
def reset_engine() -> None:
|
|
"""Drop a configured engine and rebuild from settings on the next call.
|
|
|
|
Only an engine this module built itself is closed.
|
|
"""
|
|
global _engine, _engine_is_default
|
|
if _engine is not None and _engine_is_default:
|
|
_engine.dispose()
|
|
_engine = None
|
|
_engine_is_default = False
|
|
|
|
|
|
def create_db_and_tables():
|
|
engine = get_engine()
|
|
existing_tables = inspect(engine).get_table_names()
|
|
if not existing_tables:
|
|
SQLModel.metadata.create_all(engine)
|
|
from alembic.config import Config
|
|
|
|
from alembic import command
|
|
|
|
alembic_cfg = Config(settings.alembic_config)
|
|
alembic_cfg.set_main_option("sqlalchemy.url", str(engine.url))
|
|
alembic_cfg.attributes["sqlalchemy.url"] = settings.sqlalchemy_database_url
|
|
command.stamp(alembic_cfg, "head")
|
|
logger.info("Database created and tables initialized.")
|
|
else:
|
|
logger.info(
|
|
"Database already initialized (%d tables found).", len(existing_tables)
|
|
)
|
|
|
|
|
|
def get_session():
|
|
with Session(get_engine()) as db:
|
|
yield db
|
|
|
|
|
|
def get_db_session():
|
|
return Session(get_engine())
|
|
|
|
|
|
def add_and_refresh(db: Session, obj):
|
|
db.add(obj)
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return obj
|