alembic_utils (0.7.7) duplicates problem (https://github.com/olirice/alembic_utils/issues/85)

_utils.py

import pkgutil

import sqlparse


def read_template(path: str) -> str:
    # could be changed on `importlib_resources` after Python 3.9+
    # ref: https://stackoverflow.com/a/58941536
    template_path = f"templates/{path}"
    content = pkgutil.get_data(__name__, template_path)
    if not content:
        raise RuntimeError(f"could not get file data for {template_path}")
    return content.decode()


def read_sql_template(path: str, **kwrags) -> str:
    template = read_template(path)
    sql = template.format(**kwrags)

    # TODO: alembic_utils doesn't support comments in sql string, so we need to strip
    #  comments manually
    return sqlparse.format(sql, strip_comments=True, strip_whitespace=True).strip()

env.py

import logging.config
from typing import Optional

import alembic
import sqlalchemy
import sqlalchemy.pool


# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
ALEMBIC_CONFIG = alembic.context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if ALEMBIC_CONFIG.config_file_name is not None:
    logging.config.fileConfig(ALEMBIC_CONFIG.config_file_name)

# these imports help us to use alembic without coupling to the codebase
try:
    # for 'autogenerate'
    from app import db

    target_metadata: Optional[sqlalchemy.MetaData]
    target_metadata = db.METADATA
except ModuleNotFoundError:
    target_metadata = None

try:
    # for 'autogenerate'
    import alembic_utils.replaceable_entity

    from . import functions
    f# rom schemas.database import triggers

    replaceable_entities = []
    replaceable_entities.extend(functions.ALL_FUNCTIONS)
    # replaceable_entities.extend(triggers.ALL_TRIGGERS)

    alembic_utils.replaceable_entity.register_entities(replaceable_entities)
except ModuleNotFoundError:
    pass


def get_url():
    import os

    # mostly used by CI/CD
    dsn = os.getenv("POSTGRES_DSN", None)
    if dsn:
        return dsn

    user = os.getenv("POSTGRES_USER", "briquette")
    password = os.getenv("POSTGRES_PASSWORD", "briquette")
    server = os.getenv("POSTGRES_SERVER", "briquette_database")
    port = os.getenv("POSTGRES_PORT", None)
    if port:
        server += ":" + port
    database = os.getenv("POSTGRES_DATABASE", "briquette")
    return f"postgresql+psycopg2://{user}:{password}@{server}/{database}"


def run_migrations_offline():
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

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

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


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

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

    """
    configuration = ALEMBIC_CONFIG.get_section(ALEMBIC_CONFIG.config_ini_section, {})
    configuration["sqlalchemy.url"] = get_url()

    connectable = sqlalchemy.engine_from_config(
        configuration,
        prefix="sqlalchemy.",
        poolclass=sqlalchemy.pool.NullPool,
    )

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

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


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

functions.py

from alembic_utils.pg_function import PGFunction

from . import _utils


ALL_FUNCTIONS = [
    PGFunction.from_sql(_utils.read_sql_template("functions/copy_subtree.sql")),
    PGFunction.from_sql(_utils.read_sql_template("functions/delete_subtree.sql")),
    PGFunction.from_sql(_utils.read_sql_template("functions/select_subtree.sql")),
]