Alembic 自定义迁移代码不起作用 bigquery

Question

我在 alembic 迁移代码中尝试此查询

def get_my_table_id_list(connection):
    duplicate_id_list = connection.execute(
        """
        SELECT
            a.id
        FROM
            my table AS a
        WHERE
            a.created_datetime > (
                SELECT
                    MIN(b.created_datetime)
                FROM
                    map_direction AS b
                WHERE
                    < condition >)
        LIMIT 1000
        """
    ).fetchall()

    return duplicate_id_list


def delete_my_table(connection, id_list):
    test = connection.execute(
        f"""
        DELETE FROM map_direction
        WHERE my table.id IN ({", ".join(id_list)})
        """
    )

    print(test.rowcount)


def check_and_delete_duplicate_my_tablle(connection):
    duplicate_id_list = get_duplicate_id_list(connection)

    if len(duplicate_id_list) < 1:
        return True
    else:
        duplicate_id_list = list(map(lambda _id: str(_id[0]), duplicate_id_list))

        delete_my_table(connection, duplicate_id_list)

        check_and_delete_duplicate_my_table(connection)

    return False

在我的迁移代码中使用这个函数并运行flask db upgrade or alembic upgrade +1

我的数据库是psql，我的表所有行：5,000,000 个非常大的表

如果我的数据库行少于 100,000，这些函数对我有用。 但是我的桌子非常非常大！这个功能对我不起作用。

我该如何解决这个问题？

Answer 1

def get_my_table_id_list(connection):
    duplicate_id_list = connection.execute(
        """
        SELECT
            a.id
        FROM
            my table AS a
        WHERE
            a.id > (
                SELECT
                    MIN(b.id)
                FROM
                    map_direction AS b
                WHERE
                    < condition >)
        LIMIT 1000
        """
    ).fetchall()

    return duplicate_id_list

............ create_datetime 重复。