fix: use correct interpolation in alembic url

Author: ankaisen

src/memu/database/postgres/migration.py

@@ -1,13 +1,12 @@
 from __future__ import annotations
 
 import logging
+import re
 from pathlib import Path
 from typing import Any, Literal
 
 from sqlalchemy import create_engine, inspect, text
 
-from memu.database.postgres.schema import get_metadata
-
 try:  # Optional dependency for Postgres backend
     from alembic import command
     from alembic.config import Config as AlembicConfig
@@ -18,12 +17,17 @@
 logger = logging.getLogger(__name__)
 
 DDLMode = Literal["create", "validate"]
+_UNESCAPED_CONFIGPARSER_PERCENT = re.compile(r"(?<!%)%(?!%)")
+
+
+def _escape_for_config_parser(value: str) -> str:
+    return _UNESCAPED_CONFIGPARSER_PERCENT.sub("%%", value)
 
 
 def make_alembic_config(*, dsn: str, scope_model: type[Any]) -> AlembicConfig:
     cfg = AlembicConfig()
     cfg.set_main_option("script_location", str(Path(__file__).with_name("migrations")))
-    cfg.set_main_option("sqlalchemy.url", dsn)
+    cfg.set_main_option("sqlalchemy.url", _escape_for_config_parser(dsn))
     cfg.attributes["scope_model"] = scope_model
     return cfg
 
@@ -37,6 +41,8 @@ def run_migrations(*, dsn: str, scope_model: type[Any], ddl_mode: DDLMode = "cre
         scope_model: User scope model for scoped tables
         ddl_mode: "create" to create missing tables, "validate" to only check schema
     """
+    from memu.database.postgres.schema import get_metadata
+
     metadata = get_metadata(scope_model)
     engine = create_engine(dsn)
 

tests/test_postgres_migration_config.py

@@ -0,0 +1,31 @@
+import importlib.util
+from pathlib import Path
+
+from pydantic import BaseModel
+
+
+def _load_migration_module():
+    module_path = Path(__file__).resolve().parents[1] / "src/memu/database/postgres/migration.py"
+    spec = importlib.util.spec_from_file_location("memu_postgres_migration", module_path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class _ScopeModel(BaseModel):
+    user_id: str
+
+
+def test_make_alembic_config_escapes_percent_encoded_dsn() -> None:
+    migration = _load_migration_module()
+
+    dsn = "postgresql+psycopg://postgres:%40%23%24%25%5E%26%2A%28%29password@host.docker.internal:5432/memu_dev"
+
+    cfg = migration.make_alembic_config(dsn=dsn, scope_model=_ScopeModel)
+    raw_value = cfg.file_config.get(cfg.config_ini_section, "sqlalchemy.url", raw=True)
+
+    assert cfg.get_main_option("sqlalchemy.url") == dsn
+    assert "%%40%%23%%24%%25%%5E%%26%%2A%%28%%29password" in raw_value
+    assert raw_value != dsn