sqlalchemy / alembic

A database migrations tool for SQLAlchemy.
MIT License
2.76k stars 241 forks source link

Alembic Config does work properly with sqlalchemy.URL #1489

Closed aqeelat closed 4 months ago

aqeelat commented 4 months ago

Describe the bug Given:

from sqlalchemy import URL

DB_URL = URL.create(
        "postgresql",
        username="user",
        password="p@ssword!",
        host="localhost",
        port="5432",
        database="user_db",
    )
  1. you cannot runconfig.set_main_option("sqlalchemy.url", DB_URL)) because configparser expects the values to be strings.
  2. running str(DB_URL) would not work because passwords would be hidden. Which means you have to do DB_URL.render_as_string(True)
  3. Step 2 would not work if you have any escaped characters because URL uses urllib.parse which converts characters such as @ to %40, which then causes configparser to raise an invalid interpolation syntax. You need to escape each % with another %.

Expected behavior

It should work, or a workaround be documented, especially that context.configure has signature url: Union[str, URL, None] = None

To Reproduce See above.

Versions.

Additional context Our current workaround is by not passing sqlalchemy.url to alembic.Config at all.

from logging.config import fileConfig

from alembic import context
from app.database import Base
from settings import DB_URL
from sqlalchemy import engine_from_config
from sqlalchemy import pool

config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata

def run_migrations_offline() -> None:

    context.configure(
        url=DB_URL,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    configuration = config.get_section(config.config_ini_section, {})
    configuration["sqlalchemy.url"] = DB_URL
    connectable = engine_from_config(
        configuration=configuration,
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)

        with context.begin_transaction():
            context.run_migrations()

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

Have a nice day!