为什么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

上次迁移(Feed:Countries)对前一个(Feed:Continents)提供的表运行查询.如果我有大陆表创建和馈送,脚本应该工作.但事实并非如此.为什么我必须在此之间停止迁移过程以在另一个命令中重新启动它？我真的不明白这一点.是否有一些命令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)

非常感谢,

更新1:最后两次迁移的升级方法

正如@Miguel在评论中所说,这里有最后两次迁移的升级方法:

饲料:大陆

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:Countries提供国家/地区表.但它必须查询大陆表,以便在国家和大陆之间建立适当的联系.

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

我在Reddit问了同样的问题,themathemagician说:

我之前遇到过这个问题,问题是迁移不是单独执行的,而是将alembic批处理所有这些(或者所有需要运行的程序),然后执行SQL.这意味着在上次迁移尝试运行时,表实际上并不存在,因此您无法实际进行查询.干

from alembic import op

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

这不是最优雅的解决方案(对于Postgres来说,命令可能与其他dbs不同),但它对我有用.此外,这实际上不是Flask-Migrate的问题,而是与alembic的问题,所以如果你想谷歌获取更多信息,请搜索alembic.Flask-Migrate只是一个围绕alembic的包装器,可以轻松地与Flask-Script一起使用.

Answer 1

正如@dhemathemagician on reddit所示,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
                )
    # ...

另请注意,如果您还计划运行脱机迁移,则需要以相同方式修复configure()呼叫run_migrations_offline().

尝试一下,让我知道它是否解决了这个问题.