add database migration tooling with alembic

src/wce_crm/README.md

@@ -5,3 +5,4 @@
 - DOPT_DB_CRM: path to CRM database, relative to base path
 - DOPT_DB_MAIN: path to main database, relative to base path
 - DOPT_PATH_LOGGING: path to logging folder, relative to base path
+- DOPT_ALEMBIC_BASE: path to all relevant alembic file

src/wce_crm/constants.py

@@ -24,3 +24,4 @@ class Config:
     )
     PATH_LOGGING: Path = BASE_PATH / os.getenv("DOPT_PATH_LOGGING", "data/d-opt.log")
     LOG_FILENAME: str = "dopt.log"
+    ALEMBIC_PATH: Path = BASE_PATH / os.getenv("DOPT_ALEMBIC_BASE", "python/alembic")

src/wce_crm/db.py

@@ -13,6 +13,7 @@ from wce_crm import constants
 from wce_crm import types as t
 
 
+# // declarations
 class SafeDateTime(TypeDecorator):
     """Cleans non-standard ISO strings before parsing."""
 
@@ -57,8 +58,8 @@ class UTCDateTime(TypeDecorator):
         return value
 
 
-md_crm = sql.MetaData()
-md_main = sql.MetaData()
+MD_CRM = sql.MetaData()
+MD_MAIN = sql.MetaData()
 ENGINE = sql.create_engine(f"sqlite:///{constants.Config.DB_PATH_MAIN}")
 
 # ---------- OLD "Kontaktliste" ----------
@@ -141,7 +142,7 @@ ENGINE = sql.create_engine(f"sqlite:///{constants.Config.DB_PATH_MAIN}")
 
 ext_crm_master: sql.Table = Table(
     "Master",
-    md_crm,
+    MD_CRM,
     Column("ma_id", sql.Integer, nullable=False, unique=True),
     Column("wce_id", sql.ForeignKey("Nutzer.wce_id")),
     Column("ma_unternehmensname", sql.Text, nullable=True),
@@ -218,7 +219,7 @@ DF_CRM_MASTER = get_ext_crm_master(constants.Config.DB_PATH_CRM)
 
 ext_crm_nutzer: sql.Table = Table(
     "Nutzer",
-    md_crm,
+    MD_CRM,
     Column("wce_id", sql.Integer, nullable=False, unique=True),
     Column("wce_name", sql.Text, nullable=True),
     Column("wce_vorname", sql.Text, nullable=True),
@@ -246,7 +247,7 @@ ext_crm_nutzer_schema: t.PolarsSchema = {
 
 ext_crm_contact_person: sql.Table = Table(
     "Ansprechpartner",
-    md_crm,
+    MD_CRM,
     Column("an_id", sql.Integer, nullable=False, unique=True),
     Column("ma_id", sql.ForeignKey("Master.ma_id")),
     Column("wce_id", sql.ForeignKey("Nutzer.wce_id")),
@@ -325,7 +326,7 @@ DF_CONTACT_PERSON = get_ext_crm_contact_person(constants.Config.DB_PATH_CRM)
 
 grunderfassung: sql.Table = Table(
     "grunderfassung",
-    md_main,
+    MD_MAIN,
     Column(
         "erfassung_id",
         sql.Integer,
@@ -398,4 +399,4 @@ grunderfassung: sql.Table = Table(
     Column("WeitereInfos__WI_meldung_institution", sql.Text, nullable=True),
 )
 
-md_main.create_all(ENGINE)
+MD_MAIN.create_all(ENGINE)

src/wce_crm/db_migration.py

@@ -0,0 +1,36 @@
+from alembic import command
+from alembic.config import Config
+
+from wce_crm import constants
+from wce_crm.logging import logger_base as logger
+
+
+def check_and_run_migrations() -> None:
+    logger.info("[DB migration - Alembic] Start...")
+    alembic_base_folder = constants.Config.ALEMBIC_PATH
+    alembic_ini_file = alembic_base_folder / "alembic.ini"
+    alembic_working_dir = alembic_base_folder / "alembic"
+
+    assert alembic_ini_file.exists(), "Alembic INI file not found"
+    assert alembic_working_dir.exists(), "Alembic working dir not found"
+    assert alembic_working_dir.is_dir(), "Alembic working dir is not a directory"
+
+    logger.info("[DB migration - Alembic] Initialise config...")
+    alembic_cfg = Config(alembic_ini_file)
+    alembic_cfg.set_main_option("script_location", str(alembic_working_dir))
+
+    logger.info("[DB migration - Alembic] Run...")
+    try:
+        command.upgrade(alembic_cfg, "head")
+        logger.info("[DB migration - Alembic] Database up-to-date.")
+    except Exception as err:
+        logger.info(
+            "[DB migration - Alembic] An error occurred during the migration:\n%s",
+            err,
+            stack_info=True,
+        )
+
+
+if __name__ == "__main__":
+    # start migration process before app start
+    check_and_run_migrations()