mysql password supports special characters

dandelion/alembic/env.py

@@ -23,6 +23,7 @@
 import dandelion.conf
 from dandelion import constants, version
 from dandelion.db.base import Base  # noqa
+from dandelion.db.session import connection_database
 
 CONF: cfg = dandelion.conf.CONF
 
@@ -49,6 +50,7 @@
 
 target_metadata = Base.metadata
 
+
 # other values from the config, defined by the needs of env.py,
 # can be acquired:
 # my_important_option = config.get_main_option("my_important_option")
@@ -67,8 +69,9 @@ def run_migrations_offline():
     script output.
 
     """
+    connection = connection_database()
     context.configure(
-        url=CONF.database.connection,
+        url=connection,
         target_metadata=target_metadata,
         literal_binds=True,
         dialect_opts={"paramstyle": "named"},
@@ -85,7 +88,8 @@ def run_migrations_online():
     and associate a connection with the context.
 
     """
-    engine = create_engine(CONF.database.connection, poolclass=pool.NullPool)
+    connection = connection_database()
+    engine = create_engine(connection, poolclass=pool.NullPool)
 
     with engine.connect() as connection:
         context.configure(connection=connection, target_metadata=target_metadata)

dandelion/db/session.py

@@ -16,6 +16,7 @@
 
 from logging import LoggerAdapter
 from typing import Dict
+from urllib.parse import quote_plus
 
 from oslo_config import cfg
 from oslo_log import log
@@ -25,7 +26,6 @@
 CONF = cfg.CONF
 LOG: LoggerAdapter = log.getLogger(__name__)
 
-
 DB_SESSION_LOCAL: Session
 
 
@@ -40,9 +40,20 @@ def setup_db() -> None:
         engine_cfg: Dict[str, int] = {}
         engine_cfg["pool_size"] = CONF.database.max_pool_size
         engine_cfg["max_overflow"] = CONF.database.max_overflow
-        engine = create_engine(CONF.database.connection, pool_pre_ping=True, **engine_cfg)
+
+        connection = connection_database()
+
+        engine = create_engine(connection, pool_pre_ping=True, **engine_cfg)
 
     global DB_SESSION_LOCAL
     DB_SESSION_LOCAL = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
     LOG.info("DB setup complete")
+
+
+def connection_database() -> str:
+    connection_list = CONF.database.connection.split(":")
+    index = connection_list[2].rfind("@")
+    connection_list[2] = quote_plus(connection_list[2][:index]) + connection_list[2][index:]
+    connection = ":".join(connection_list)
+    return connection