Flask-SQLAlchemy Alembic迁移

问题描述

我正在尝试使用Flask-SQLAlchemy和alembic在postgresDB上进行数据库迁移。

models.py

import datetime

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()


class TestOne(db.Model):
    __tablename__ = 'test_one'
    __bind_key__ = 'tester'
    __table_args__ = {"schema": "schema1"}

    id = db.Column(db.Integer,primary_key=True)
    column1 = db.Column(db.String(50),unique=True,nullable=False)
    timestamp = db.Column(db.DateTime,default=datetime.datetime.utcnow)

    def __str__(self):
        return f"<TestOne {self.column1}>"


class TestTwo(db.Model):
    __tablename__ = 'test_two'
    __bind_key__ = 'tester'

    id = db.Column(db.Integer,primary_key=True)
    code = db.Column("code",db.String(50),index=True,nullable=False)
    f_date = db.Column(db.DateTime,nullable=False)
    t_date = db.Column(db.Date,nullable=False)
    modified = db.Column("modified",db.Boolean,nullable=False)

    __table_args__ = (
        db.Index(
            "ix_test_two_code_modified",code,modified,postgresql_where=(modified.is_(True))
        ),{"schema": "public"},)

    def __str__(self):
        return f"<TestTwo {self.code}>"

数据库设置

def setup_database(app):
    from .models import db

    db.init_app(app)
    migrate.init_app(app,db,include_schema=True)
    manager = Manager(app)
    manager.add_command('db',MigrateCommand)

    with app.app_context():
        db.create_all(app=app,bind='tester')
        return app

并按照文档中指定的步骤进行操作

flask db init
flask db migrate -m "initial"
flask db upgrade

烧瓶db migration -m“初始迁移” 在migrartion> versions目录中创建initial.py文件，并在postgresDB中创建表

initial.py

"""initial

Revision ID: 65bc44e61e98
Revises: 
Create Date: 2020-08-11 23:17:08.664515

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers,used by Alembic.
revision = '65bc44e61e98'
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('test_two',sa.Column('id',sa.Integer(),nullable=False),sa.Column('code',sa.String(length=50),sa.Column('f_date',sa.DateTime(),sa.Column('t_date',sa.Date(),sa.Column('modified',sa.Boolean(),sa.PrimaryKeyConstraint('id'),schema='public'
    )
    op.create_index(op.f('ix_public_test_two_code'),'test_two',['code'],unique=False,schema='public')
    op.create_index(op.f('ix_public_test_two_modified'),['modified'],schema='public')
    op.create_index('ix_test_two_code_modified',['code','modified'],schema='public',postgresql_where=sa.text('modified IS 1'))
    op.create_table('test_one',sa.Column('column1',sa.Column('timestamp',nullable=True),sa.UniqueConstraint('column1'),schema='schema1'
    )
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('test_one',schema='schema1')
    op.drop_index('ix_test_two_code_modified',table_name='test_two',schema='public')
    op.drop_index(op.f('ix_public_test_two_modified'),schema='public')
    op.drop_index(op.f('ix_public_test_two_code'),schema='public')
    op.drop_table('test_two',schema='public')
    # ### end Alembic commands ###

但是烧瓶数据库升级失败并显示错误

    cursor.execute(statement,parameters)
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) unknown database public
[SQL:
CREATE TABLE public.test_two (
        id INTEGER NOT NULL,code VARCHAR(50) NOT NULL,f_date DATETIME NOT NULL,t_date DATE NOT NULL,modified BOOLEAN NOT NULL,PRIMARY KEY (id),CHECK (modified IN (0,1))
)

]
(Background on this error at: http://sqlalche.me/e/13/e3q8)

不确定为什么要考虑将架构公开为“数据库”，或者甚至将include_schema参数添加到True以使用多个架构，也不确定是什么问题？

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）