How to avoid DuplicateTableError when upgrading alembic migrations using async?

[jalvespinto]

I am trying to upgrade a db with alembic where I use asyncpg and I am getting the error

sqlalchemy.exc.ProgrammingError: (sqlalchemy.dialects.postgresql.asyncpg.ProgrammingError) <class 'asyncpg.exceptions.DuplicateTableError'>: relation "receipts_houses_association" already exists

I tried using this solution, but I guess inspector does not work with async code. Any ideas of how to do it?

[CaselIT]

Hi,

Inside the alembic upgrade or downgrade you can use the inspector as in that example.
Alembic runs in sync mode even with async dbs, so you can run everything as if it were a normal sync connection. SQLAlchemy then takes case of running the actual sql query as async

[jalvespinto]

Hi @CaselIT.

I still cannot get it to work. I have the versions bellow until now. On the second one I get the error. If I delete the association table then I get an error when creating the items table, because the receipts tables does not exist.

"""initial iam

Revision ID: 644f7e0a7b57
Revises: 
Create Date: 2023-05-24 20:21:48.317697

"""
import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic.
revision = '644f7e0a7b57'
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute('CREATE SCHEMA IF NOT EXISTS iam')
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('roles',
    sa.Column('name', sa.String(), nullable=False),
    sa.Column('is_default', sa.Boolean(), nullable=False),
    sa.Column('permissions', sa.Integer(), nullable=False),
    sa.PrimaryKeyConstraint('name', name=op.f('pk_roles')),
    schema='iam'
    )
    op.create_table('users',
    sa.Column('id', sa.String(), nullable=False),
    sa.Column('user_name', sa.String(), nullable=False),
    sa.Column('email', sa.String(), nullable=False),
    sa.Column('hashed_password', sa.String(), nullable=False),
    sa.Column('confirmed', sa.Boolean(), nullable=False),
    sa.Column('active', sa.Boolean(), nullable=False),
    sa.Column('role_id', sa.String(), nullable=False),
    sa.Column('version', sa.Integer(), nullable=False),
    sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
    sa.ForeignKeyConstraint(['role_id'], ['iam.roles.name'], name=op.f('fk_users_role_id_roles')),
    sa.PrimaryKeyConstraint('id', name=op.f('pk_users')),
    schema='iam'
    )
    op.create_index(op.f('ix_iam_users_email'), 'users', ['email'], unique=True, schema='iam')
    op.create_index(op.f('ix_iam_users_user_name'), 'users', ['user_name'], unique=True, schema='iam')
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index(op.f('ix_iam_users_user_name'), table_name='users', schema='iam')
    op.drop_index(op.f('ix_iam_users_email'), table_name='users', schema='iam')
    op.drop_table('users', schema='iam')
    op.drop_table('roles', schema='iam')
    # ### end Alembic commands ###
    op.execute('DROP SCHEMA IF EXISTS iam')

"""initial receipt tracker

Revision ID: ed54121da6e2
Revises: 644f7e0a7b57
Create Date: 2023-05-24 20:25:51.923651

"""
import sqlalchemy as sa
from alembic import op
from sqlalchemy.engine.reflection import Inspector

# revision identifiers, used by Alembic.
revision = "ed54121da6e2"
# down_revision = None
down_revision = "644f7e0a7b57"
branch_labels = None
depends_on = None

def upgrade() -> None:
    conn = op.get_bind()
    inspector = Inspector.from_engine(conn)
    op.execute("CREATE SCHEMA IF NOT EXISTS receipt_tracker")
    # ### commands auto generated by Alembic - please adjust! ###
    print(inspector.get_table_names(schema="receipt_tracker"))
    if "houses" in inspector.get_table_names(schema="receipt_tracker"):
        op.create_table(
            "houses",
            sa.Column("id", sa.String(), nullable=False),
            sa.PrimaryKeyConstraint("id", name=op.f("pk_houses")),
            schema="receipt_tracker",
        )
    print(inspector.get_table_names(schema="receipt_tracker"))
    if "sellers" in inspector.get_table_names(schema="receipt_tracker"):
        op.create_table(
            "sellers",
            sa.Column("id", sa.String(), nullable=False),
            sa.Column("name", sa.String(), nullable=False),
            sa.Column("state_registration", sa.String(), nullable=False),
            sa.Column("street", sa.String(), nullable=False),
            sa.Column("number", sa.String(), nullable=False),
            sa.Column("zip_code", sa.String(), nullable=False),
            sa.Column("district", sa.String(), nullable=False),
            sa.Column("city", sa.String(), nullable=False),
            sa.Column("state", sa.String(), nullable=False),
            sa.Column("complement", sa.String(), nullable=True),
            sa.Column("note", sa.String(), nullable=True),
            sa.PrimaryKeyConstraint("id", name=op.f("pk_sellers")),
            schema="receipt_tracker",
        )
    print(inspector.get_table_names(schema="receipt_tracker"))
    if "receipts" in inspector.get_table_names(schema="receipt_tracker"):
        op.create_table(
            "receipts",
            sa.Column("id", sa.String(), nullable=False),
            sa.Column("qrcode", sa.String(), nullable=True),
            sa.Column("date", sa.DateTime(), nullable=True),
            sa.Column("state", sa.String(), nullable=True),
            sa.Column("seller_id", sa.String(), nullable=True),
            sa.Column("scraped", sa.Boolean(), nullable=True),
            sa.Column("discarded", sa.Boolean(), nullable=False),
            sa.Column("version", sa.Integer(), nullable=False),
            sa.ForeignKeyConstraint(
                ["seller_id"],
                ["receipt_tracker.sellers.id"],
                name=op.f("fk_receipts_seller_id_sellers"),
            ),
            sa.PrimaryKeyConstraint("id", name=op.f("pk_receipts")),
            schema="receipt_tracker",
        )
    print(inspector.get_table_names(schema="receipt_tracker"))
    if "receipts_houses_association" not in inspector.get_table_names(schema="receipt_tracker"):
        op.create_table(
            "receipts_houses_association",
            sa.Column("receipt_id", sa.String(), nullable=False),
            sa.Column("house_id", sa.String(), nullable=False),
            sa.ForeignKeyConstraint(
                ["receipt_id"],
                ["receipt_tracker.receipts.id"],
                name=op.f("fk_association_table_receipt_id_receipts"),
            ),
            sa.ForeignKeyConstraint(
                ["house_id"],
                ["receipt_tracker.houses.id"],
                name=op.f("fk_association_table_house_id_houses"),
            ),
            sa.PrimaryKeyConstraint(
                "receipt_id", "house_id", name=op.f("receipts_houses_association")
            ),
            schema="receipt_tracker",
        )
    print(inspector.get_table_names(schema="receipt_tracker"))
    if "items" not in inspector.get_table_names(schema="receipt_tracker"):
        op.create_table(
            "items",
            sa.Column("number", sa.Integer(), nullable=False),
            sa.Column("receipt_id", sa.String(), nullable=False),
            sa.Column("description", sa.String(), nullable=False),
            sa.Column("quantity", sa.Float(), nullable=False),
            sa.Column("unit", sa.String(), nullable=False),
            sa.Column("price_paid", sa.Float(), nullable=False),
            sa.Column("price_per_unit", sa.Float(), nullable=False),
            sa.Column("gross_price", sa.Float(), nullable=False),
            sa.Column("sellers_product_code", sa.String(), nullable=False),
            sa.Column("barcode", sa.String(), nullable=False),
            sa.Column("discount", sa.Float(), nullable=False),
            sa.ForeignKeyConstraint(
                ["receipt_id"],
                ["receipt_tracker.receipts.id"],
                name=op.f("fk_items_receipt_id_receipts"),
            ),
            sa.PrimaryKeyConstraint("number", "receipt_id", name=op.f("pk_items")),
            schema="receipt_tracker",
        )
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table("items", schema="receipt_tracker")
    op.drop_table("receipts_houses_association", schema="receipt_tracker")
    op.drop_table("receipts", schema="receipt_tracker")
    op.drop_table("sellers", schema="receipt_tracker")
    op.drop_table("houses", schema="receipt_tracker")
    # ### end Alembic commands ###
    op.execute("DROP SCHEMA IF EXISTS receipt_tracker")

[CaselIT]

You seem to have a few typos here:

if "houses" in inspector.get_table_names(schema="receipt_tracker"):
if "sellers" in inspector.get_table_names(schema="receipt_tracker"):
if "receipts" in inspector.get_table_names(schema="receipt_tracker"):

you should probably use not in.

Also on sqlalchemy 1.4+ you can use if not inspector.has_table('houses', schema="receipt_tracker"):

[jalvespinto]

I found out what I was doing wrong. I had some typos on the association table and also I had to op.get_bind() on every if statement. Tks for the help @CaselIT .

[CaselIT]

oh ok. great then!