为什么 Flask-Migrate 让我进行两步迁移？

Question

我正在使用 Flask、SQLAlchemy、Alembic 及其 Flask 包装器（Flask-SQLAlchemy 和 Flask-Migrate）开展一个项目。我有四个迁移：

1c5f54d4aa34 -> 4250dfa822a4 (head), Feed: Countries
312c1d408043 -> 1c5f54d4aa34, Feed: Continents
41984a51dbb2 -> 312c1d408043, Basic Structure
<base> -> 41984a51dbb2, Init Alembic

当我启动一个新的、干净的数据库并尝试运行迁移时，我收到一个错误：

vagrant@precise32:/vagrant$ python manage.py db upgrade
...
sqlalchemy.exc.ProgrammingError: (ProgrammingError) relation "continent" does not exist
...

如果我要求 Flask-Migrate 运行除最后一次以外的所有迁移，它可以工作。如果在那之后我再次运行升级命令，它就可以工作了——也就是说，它完全升级了我的数据库，而无需对代码进行任何更改：

vagrant@precise32:/vagrant$ python manage.py db upgrade 312c1d408043
INFO  [alembic.migration] Context impl PostgresqlImpl.
INFO  [alembic.migration] Will assume transactional DDL.
INFO  [alembic.migration] Running upgrade  -> 41984a51dbb2, Init Alembic
INFO  [alembic.migration] Running upgrade 41984a51dbb2 -> 312c1d408043, Basic Structure

vagrant@precise32:/vagrant$ python manage.py db upgrade
INFO  [alembic.migration] Context impl PostgresqlImpl.
INFO  [alembic.migration] Will assume transactional DDL.
INFO  [alembic.migration] Running upgrade 312c1d408043 -> 1c5f54d4aa34, Feed: Continents
INFO  [alembic.migration] Running upgrade 1c5f54d4aa34 -> 4250dfa822a4, Feed: Countries

TL;DR

最后一次迁移（供稿：国家/地区）对由前一次迁移（供稿：大陆）提供的表运行查询。如果我创建并提供了大洲表，则脚本应该可以工作。但事实并非如此。 为什么我必须停止迁移过程才能在另一个命令中重新启动它？我真的不明白这个。是 Alembic 在一系列迁移后执行的某个命令吗？有什么想法吗？

以防万一

我的模型定义如下：

class Country(db.Model):

    __tablename__ = 'country'
    id = db.Column(db.Integer, primary_key=True)
    alpha2 = db.Column(db.String(2), index=True, unique=True)
    title = db.Column(db.String(140))
    continent_id = db.Column(db.Integer, db.ForeignKey('continent.id'))
    continent = db.relationship('Continent', backref='countries')

    def __repr__(self):
        return '<Country #{}: {}>'.format(self.id, self.title)

class Continent(db.Model):

    __tablename__ = 'continent'
    id = db.Column(db.Integer, primary_key=True)
    alpha2 = db.Column(db.String(2), index=True, unique=True)
    title = db.Column(db.String(140))

    def __repr__(self):
        return '<Continent #{}: {}>'.format(self.id, self.title)

非常感谢，

UPDATE 1：最近两次迁移的升级方法

正如@Miguel 在评论中所问，这里有最后两次迁移的升级方法：

Feed：各大洲

def upgrade():
    csv_path = app.config['BASEDIR'].child('migrations', 'csv', 'en')
    csv_file = csv_path.child('continents.csv')
    with open(csv_file) as file_handler:
        csv = list(reader(file_handler))
        csv.pop(0)
        data = [{'alpha2': c[0].lower(), 'title': c[1]} for c in csv]
        op.bulk_insert(Continent.__table__, data)

Feed：国家（这取决于上次迁移时提供的表格）

def upgrade():

    # load countries iso3166.csv and build a dictionary
    csv_path = app.config['BASEDIR'].child('migrations', 'csv', 'en')
    csv_file = csv_path.child('iso3166.csv')
    countries = dict()
    with open(csv_file) as file_handler:
        csv = list(reader(file_handler))
        for c in csv:
            countries[c[0]] = c[1]

    # load countries-continents from country_continent.csv
    csv_file = csv_path.child('country_continent.csv')
    with open(csv_file) as file_handler:
        csv = list(reader(file_handler))
        country_continent = [{'country': c[0], 'continent': c[1]} for c in csv]

    # loop
    data = list()
    for item in country_continent:

        # get continent id
        continent_guess = item['continent'].lower()
        continent = Continent.query.filter_by(alpha2=continent_guess).first()

        # include country
        if continent is not None:
            country_name = countries.get(item['country'], False)
            if country_name:
                data.append({'alpha2': item['country'].lower(),
                             'title': country_name,
                             'continent_id': continent.id})

我使用的 CSV 基本上遵循以下模式：

continents.csv

...
AS, "Asia"
EU, "Europe"
NA, "North America"
...

iso3166.csv

...
CL,"Chile"
CM,"Cameroon"
CN,"China"
...

_country_continent.csv_

...
US,NA
UY,SA
UZ,AS
...

所以 Feed: Continents 提供大陆表，而 Feed: Country 提供国家表。但它必须查询大洲表才能在国家和大洲之间建立正确的链接。

更新 2：Reddit 的一些人已经提供了解释和解决方法

我问the same question on Reddit，themathemagician说：

我以前遇到过这种情况，问题是迁移没有 单独执行，而是将它们全部（或全部 其中需要运行的），然后执行 SQL。这意味着 到最后一次迁移尝试运行时，表不会 实际存在，因此您实际上无法进行查询。正在做

from alembic import op

def upgrade():
    #migration stuff
    op.execute('COMMIT')
    #run queries

这不是最优雅的解决方案（对于 Postgres， 其他数据库的命令可能不同），但它对我有用。还， 这实际上不是 Flask-Migrate 的问题，而是问题 使用 alembic，因此如果您想在 Google 上获取更多信息，请搜索 蒸馏器。 Flask-Migrate 只是一个可以工作的 alembic 的包装器 轻松使用 Flask-Script。

Answer 1

正如 reddit 上的@themathemagician 所指出的，Alembic 默认情况下在单个事务中运行所有迁移，因此取决于数据库引擎和您在迁移脚本中执行的操作，一些依赖于先前迁移中添加的内容的操作可能失败。

我自己没有尝试过，但是 Alembic 0.6.5 引入了一个 transaction_per_migration 选项，它可能会解决这个问题。这是configure() 调用env.py 的一个选项。如果您在 Flask-Migrate 创建时使用默认配置文件，那么您可以在 migrations/env.py 中修复此问题：

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

    # ...
    context.configure(
                connection=connection,
                target_metadata=target_metadata,
                transaction_per_migration=True        # <-- add this
                )
    # ...

另请注意，如果您还计划运行离线迁移，则需要以相同方式修复 run_migrations_offline() 中的 configure() 调用。

试一试，如果它解决了问题，请告诉我。