Loading database URLs and secrets in Alembic env.py

Leave sqlalchemy.url empty in alembic.ini and build the URL in env.py with URL.create() from environment variables — the URL object escapes every component, masks the password when printed, and sidesteps the configparser interpolation error a percent sign in a password otherwise causes. This guide belongs to configuring Alembic with async SQLAlchemy engines.

Quick Answer

A URL in alembic.ini is a credential in version control, and it needs one file per environment.

Before — the URL in the config file:

# alembic.ini
[alembic]
script_location = alembic
sqlalchemy.url = postgresql+asyncpg://shop:s3cr%%et@db.internal/shop

# alembic/env.py — reads whatever the file says
connectable = async_engine_from_config(
    config.get_section(config.config_ini_section, {}), prefix="sqlalchemy."
)
# configparser.InterpolationSyntaxError if the '%' is not doubled,
# and the password is in the repository either way.

After — no URL in the file, built in env.py:

# alembic.ini
[alembic]
script_location = alembic
# sqlalchemy.url is intentionally absent: env.py builds it from the environment.

# alembic/env.py (excerpt)
import asyncio
import os
from pathlib import Path

from alembic import context
from sqlalchemy import pool
from sqlalchemy.engine import URL
from sqlalchemy.ext.asyncio import create_async_engine

from shop.models import Base

config = context.config


def _password() -> str:
    """Environment variable, or a mounted secret file if one is configured."""
    path = os.environ.get("PGPASSWORD_FILE")
    if path:
        return Path(path).read_text(encoding="utf-8").strip()
    return os.environ["PGPASSWORD"]


def database_url() -> URL:
    return URL.create(
        drivername="postgresql+asyncpg",
        username=os.environ.get("MIGRATION_PGUSER", os.environ["PGUSER"]),
        password=_password(),
        host=os.environ["PGHOST"],
        port=int(os.environ.get("PGPORT", 5432)),
        database=os.environ["PGDATABASE"],
    )


async def run_async_migrations() -> None:
    connectable = create_async_engine(database_url(), poolclass=pool.NullPool)
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

URL.create() escapes each component, so a password containing %, @, / or : needs no attention, and print(url) renders the password as *** — which matters because env.py output ends up in deploy logs.

Execution Context & Async Workflow Integration

alembic.ini is read by Python's configparser, which treats % as the start of an interpolation token. A password containing a percent sign therefore has to be written %%, and a URL assembled from an environment variable and written into the config with config.set_main_option() has to be escaped the same way:

# If you must go through the config, escape for configparser:
config.set_main_option("sqlalchemy.url", str(database_url()).replace("%", "%%"))

Building the engine directly from a URL object avoids the question entirely, which is the main reason to prefer it. It also avoids the second escaping layer: a URL string needs its components percent-encoded, and URL.create() does that correctly — including for passwords that contain characters which would otherwise be read as URL syntax.

The URL object has one more property worth relying on: its repr masks the password. A string does not, so any code path that logs a URL — a startup message, an exception, a debug print in env.py — leaks the credential into log aggregation if the URL is a string and does not if it is a URL. When a string is genuinely needed, url.render_as_string(hide_password=False) makes the decision explicit at the call site.

Two settings belong on a migration engine specifically.

poolclass=NullPool is right because a migration run is one-shot: it needs one connection for a few seconds, not a pool. With a pool, a deploy that runs several migration jobs against a connection-limited database holds connections it never uses again — and the async template uses NullPool for exactly this reason.

A separate, more privileged role is worth configuring here rather than reusing the application's. Migrations need CREATE, ALTER and DROP; the application should not have them. Reading MIGRATION_PGUSER with a fallback, as above, lets the same env.py serve both cases — a migration job sets it, a test fixture does not.

Everything else in env.py is unchanged: the URL feeds create_async_engine, the connection goes through run_sync(), and revisions run synchronously on the facade. That structure is described in setting up Alembic env.py for asyncpg, and the variant where the application passes its own connection instead is in running Alembic migrations programmatically from async code.

Resolving Warnings, Errors & Common Mistakes

IAM authentication is worth spelling out, because a migration job is exactly where a long-lived password is least necessary. The token is generated from the process identity, expires in minutes, and is supplied per connection by a do_connect listener — which also means a migration that runs longer than the token's lifetime still works, because the token is only checked at login:

# alembic/env.py (excerpt)
import boto3
from sqlalchemy import event
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import create_async_engine

_rds = boto3.client("rds", region_name=os.environ["AWS_REGION"])


def _iam_token() -> str:
    return _rds.generate_db_auth_token(
        DBHostname=os.environ["PGHOST"],
        Port=int(os.environ.get("PGPORT", 5432)),
        DBUsername=os.environ["MIGRATION_PGUSER"],
        Region=os.environ["AWS_REGION"],
    )


async def run_async_migrations() -> None:
    connectable = create_async_engine(
        database_url_without_password(), poolclass=pool.NullPool,
        connect_args={"ssl": ssl.create_default_context()},
    )

    @event.listens_for(connectable.sync_engine, "do_connect")
    def _provide_token(dialect, conn_rec, cargs, cparams):
        cparams["password"] = _iam_token()

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

The full treatment, including Cloud SQL and Azure, is in using IAM database authentication with async engines.

Advanced: One env.py for Local, CI and Production

The goal worth aiming at is a single alembic.ini and a single env.py where the only difference between environments is their variables — so alembic upgrade head is the same command everywhere, and nobody can run the wrong one by passing the wrong -c.

# alembic/env.py (excerpt)
import os
from dataclasses import dataclass
from pathlib import Path

from sqlalchemy.engine import URL


@dataclass(frozen=True)
class Settings:
    host: str
    port: int
    database: str
    username: str
    password: str
    sslmode_required: bool

    @classmethod
    def from_environment(cls) -> "Settings":
        # A single DATABASE_URL wins when present: platforms like Heroku provide only that.
        single = os.environ.get("MIGRATION_DATABASE_URL") or os.environ.get("DATABASE_URL")
        if single:
            parsed = URL.create(single) if "://" not in single else make_url(single)
            return cls(
                host=parsed.host, port=parsed.port or 5432, database=parsed.database,
                username=parsed.username, password=parsed.password or "",
                sslmode_required=os.environ.get("PGSSLMODE", "require") != "disable",
            )
        return cls(
            host=os.environ["PGHOST"],
            port=int(os.environ.get("PGPORT", 5432)),
            database=os.environ["PGDATABASE"],
            username=os.environ.get("MIGRATION_PGUSER", os.environ["PGUSER"]),
            password=_password(),
            sslmode_required=os.environ.get("PGSSLMODE", "require") != "disable",
        )

    def url(self) -> URL:
        return URL.create(
            "postgresql+asyncpg",
            username=self.username, password=self.password,
            host=self.host, port=self.port, database=self.database,
        )

Accepting a single DATABASE_URL as well as discrete variables is worth the few lines: managed platforms provide one combined URL, while Kubernetes deployments usually provide parts, and a migration job should work under both without a second env.py.

Two guardrails are worth adding to that function.

Fail loudly on a missing variable, with a message that names it. os.environ["PGHOST"] already does, and it is much better than defaulting to localhost — which is how a migration ends up being applied to a developer's database instead of staging.

Print where you are about to migrate, without the password. One line at the start of env.py prevents the worst class of mistake:

settings = Settings.from_environment()
print(f"alembic: {settings.username}@{settings.host}:{settings.port}/{settings.database}")

Because it prints the components rather than the URL, there is no way for the password to appear. During an incident, that line in the deploy log answers "which database did this actually run against" immediately.

Finally, keep the script_location absolute relative to env.py rather than to the working directory, so the command works from anywhere:

config.set_main_option("script_location", str(Path(__file__).resolve().parent))

Local Development Without Secrets in Files

The reason URLs end up committed is usually convenience: a developer wants alembic upgrade head to work without exporting five variables. That can be arranged without putting anything sensitive in the repository.

A .env file that is not committed. python-dotenv loads it in env.py, and the file is listed in .gitignore. Loading it only when it exists means production is unaffected:

# alembic/env.py (excerpt)
from pathlib import Path

try:
    from dotenv import load_dotenv
except ImportError:
    load_dotenv = None

if load_dotenv is not None:
    env_file = Path(__file__).resolve().parent.parent / ".env"
    if env_file.exists():
        load_dotenv(env_file, override=False)     # never override real environment values

override=False is the important flag: a variable already set by the platform wins, so a stray .env in a container image cannot redirect a production migration.

Defaults only for values that cannot cause damage. A local host and port are safe to default; a database name and credentials are not. The asymmetry is deliberate — a missing PGDATABASE should stop the command, while a missing PGPORT can reasonably assume 5432.

A committed template. .env.example documents every variable the migration needs, with placeholder values, so a new developer knows what to set and CI can be checked against the same list.

For tests, none of this applies: the test fixture creates its own database and passes the URL directly, as in running Alembic migrations programmatically from async code. That is the reason to keep URL construction in a function rather than at module level in env.py: a test can then build its own Settings without the environment being involved at all.

One last habit worth adopting: check that no credential has ever been committed, rather than assuming. A scan of the repository history for connection strings — git log -p -S "postgresql://" -- alembic.ini — takes a minute, and if it finds something, the password has to be rotated regardless of whether the file still contains it. Removing a secret from the working tree does not remove it from the history, and a repository's history is as public as the repository.

Frequently Asked Questions

Related

• Configuring Alembic with Async SQLAlchemy Engines — The parent guide: async env.py structure and options.
• Setting up Alembic env.py for asyncpg — The env.py this configuration slots into.
• Running Alembic across multiple databases and tenant schemas — When one env.py has to reach several targets.
• Using IAM database authentication with async engines — Short-lived credentials instead of a stored password.