python - alembic create_table 使用 declarative_base 派生对象

Question

我有一个 Alchemy ORM 对象:

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()


class MyORM(Base):
    id = Column(Integer, primary_key=True)
    name = Column(String(128), unique=True, nullable=False)

当使用 alembic 创建表时，我执行以下操作:

def upgrade():
    op.create_table(
        'myorm',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('name', sa.String(128), nullable=False),
    )

问题:有没有办法使用MyORM类来创建表？像这样:

def upgrade():
    op.create_table(
        'myorm',
        sa.BaseObject(MyORM)
    )

Answer 1

这正是 Alembic 迁移试图避免的。如果您将迁移与模型的当前状态相关联，那么这将不是一致的升级路径。

您可以在迁移中使用声明式来创建表和迁移数据，但不能更改。您将不得不重新创建与应用程序定义分开的定义。如果您想进行数据迁移并且更熟悉 ORM 查询而不是核心查询，这会很有用。

这是一个示例迁移，它使用声明式创建具有多对多关系的 Foo 和 Bar 模型，创建表并插入一些数据。

"""declarative

Revision ID: 169ad57156f0
Revises: 29b4c2bfce6d
Create Date: 2014-06-25 09:00:06.784170
"""

revision = '169ad57156f0'
down_revision = '29b4c2bfce6d'

from alembic import op
import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

Session = sessionmaker()
Base = declarative_base()


class Foo(Base):
    __tablename__ = 'foo'

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String, nullable=False, unique=True)


class Bar(Base):
    __tablename__ = 'bar'

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String, nullable=False, unique=True)
    foos = relationship(Foo, lambda: foo_bar, backref='bars')


foo_bar = sa.Table(
    'foo_bar', Base.metadata,
    sa.Column('foo_id', sa.Integer, sa.ForeignKey('foo.id'), primary_key=True),
    sa.Column('bar_id', sa.Integer, sa.ForeignKey('bar.id'), primary_key=True)
)

def upgrade():
    bind = op.get_bind()

    Base.metadata.create_all(bind=bind)

    session = Session(bind=bind)
    session._model_changes = False  # if you are using Flask-SQLAlchemy, this works around a bug

    f1 = Foo(name='f1')
    f2 = Foo(name='f2')
    b1 = Bar(name='b1')
    b2 = Bar(name='b2')

    f1.bars = [b1, b2]
    b2.foos.append(f2)

    session.add_all([f1, f2, b1, b2])
    session.commit()


def downgrade():
    bind = op.get_bind()

    # in this case all we need to do is drop the tables
    # Base.metadata.drop_all(bind=bind)

    # but we could also delete data
    session = Session(bind=bind)
    session._model_changes = False  # if you are using Flask-SQLAlchemy, this works around a bug

    b1 = session.query(Bar).filter_by(name='b1').one()

    session.delete(b1)
    session.commit()