SqlAlchemy Alembic数据库升级与降级简易教程

郎正平

2023-12-01

前言

通常我们会将我们的代码放入到某个VCS（版本控制系统）中，进行可追溯的版本管理。一个项目除了代码，通常还会有一个数据库，这个数据库可能会随着项目的演进发生变化，甚至需要可以回滚到过去的某个状态，于是一些工具将数据库的版本化也纳入了管理。

Alembic是Sqlalchemy的作者实现的一个数据库版本化管理工具，它可以对基于Sqlalchemy的Model与数据库之间的历史关系进行版本化的维护。

开始

Auto Generating Migrations

安装

pip install alembic

初始化

alembic init alembic

.
├── alembic 
│   ├── env.py            # 配置DataBase.metadata
│   ├── README
│   ├── script.py.mako    # 迁移脚本生成模版
│   └── versions          # 版本更新(迁移脚本)的内容目录
└── alembic.ini           # 配置数据库

数据库模板类

# ./database.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqldb://user_name:password@localhost:3306/database_name'
db = SQLAlchemy(app)

    ...emmmm...

配置

alembic.ini
- 改sqlalchemy.url = driver://user:pass@localhost/dbname
- 为mysql+mysqldb://user_name:password@localhost:3306/database_name
alembic/env.py
- 改target_metadata = None为以下内容：

import sys                                             
from database import db              

# 这个路径要改为数据库模板类所在的目录下
# dirname(dirname(abspath(__file__))) -> ./alembic
sys.path.append(os.path.dirname(os.path.abspath(__file__)) + "/../")
from db import db

target_metadata = db.metadata

使用

这东西好处就是可以像Git一样升级和降级数据库。

这个工具是按照数据库来判断你是否要对数据库进行升级。因此，参照物是数据库表，而变化的是你写的代码。

创建数据库版本[手写]

alembic revision -m "注释"
- -m：注释
- 缺点：手写……

跑完之后会在./alembic/versions生成脚本。

alembic
├── env.py
├── README
├── script.py.mako
└── versions
    └── 6f406f586bbb_注释.py

脚本内容如下所示：

"""注释

Revision ID: 6f406f586bbb
Revises: 8b6caa4ac725
Create Date: 2019-05-14 20:06:39.940623

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '6f406f586bbb'
down_revision = '8b6caa4ac725'
branch_labels = None
depends_on = None


def upgrade():
    pass


def downgrade():
    pass

创建一个新表

Operations的文档：Operation Reference


...省略...

def upgrade():
    op.create_table(
        'account',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('name', sa.String(50), nullable=False),
        sa.Column('description', sa.Unicode(200)),
    )

def downgrade():
    op.drop_table('account')

修改表

"""Add a column

Revision ID: ae1027a6acf
Revises: 1975ea83b712
Create Date: 2011-11-08 12:37:36.714947

"""

# revision identifiers, used by Alembic.
revision = 'ae1027a6acf'
down_revision = '1975ea83b712'

from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column('account', sa.Column('last_transaction_date', sa.DateTime))

def downgrade():
    op.drop_column('account', 'last_transaction_date')

总而言之，你可以用Alembic来创建和修改表的结构。

创建数据库版本[自动]

alembic revision --autogenerate -m "initdb"
- 自动生成数据库版本
- 说明:当且仅当数据库为空的时候才会自动根据创建的Model创建version版本。

没有数据库表

没有数据库的时候，Alembic会根据模型创建，命令行执行的结果：

INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table '...'
INFO  [alembic.autogenerate.compare] Detected added table '...'
INFO  [alembic.autogenerate.compare] Detected added table '...'
INFO  [alembic.autogenerate.compare] Detected added table '...'
INFO  [alembic.autogenerate.compare] Detected added table '...'
  Generating /.../project_name/alembic/versions/5e2d2560b497_init_db.py ... done

看看文件内容：

"""init db

Revision ID: 5e2d2560b497
Revises: 
Create Date: 2019-05-15 11:11:14.684351

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '5e2d2560b497'
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('...',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('...', sa.String(length=50), nullable=False),
    sa.Column('...', sa.DateTime(), nullable=True),
    sa.Column('...', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )

    ......
    
    op.create_table('...',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('...', sa.String(length=50), nullable=False),
    sa.Column('...', sa.DateTime(), nullable=True),
    sa.Column('...', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('...')

    ......

    op.drop_table('...')
    # ### end Alembic commands ###

有数据库表

执行后会在./alembic/versions生成脚本。

alembic
├── env.py
├── README
├── script.py.mako
└── versions
    └── 8b6caa4ac725_init_db.py

脚本内容如下所示：

"""init db

Revision ID: 8b6caa4ac725
Revises: 
Create Date: 2019-05-14 20:02:46.802136

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '8b6caa4ac725'
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    ...这里我省略了，都是对表修改操作...
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    ...这里我省略了，都是对表修改操作...
    # ### end Alembic commands ###

自动写完了代码，优秀！╮(╯▽╰)╭

有表和没表的区别

Alembic是根据数据库与代码中的Model的定义进行比对，来生成对应的版本号的。（Model可以看下一张的示例）

要是创建了数据库表，而当前的Model并没有任何修改，提交之后upgrade和downgrade是为空的。

加入没有创建数据库表，那么就根据Model中的将创建表和删除表的操作分别在upgrade和downgrade中编写出来。

示例

Model定义：

class Test(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    test1 = db.Column(db.String(50), unique=False, nullable=False)
    test2 = db.Column(db.String(120), unique=True, nullable=False)
    create_at = db.Column(db.DateTime, unique=False, nullable=True, default=datetime.utcnow)
    update_at = db.Column(db.DateTime, unique=False, nullable=True, default=datetime.utcnow)

初始化：alembic init alembic
配置（参照前几章）
初始化第一个版本alembic revision --autogenerate -m "init db"此时数据库表没有创建

INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'test'
  Generating /.../project_name/alembic/versions/3b63c8e4c302_init_db.py ... done

文件./alembic/versions/3b63c8e4c302_init_db.py

"""init db

Revision ID: 3b63c8e4c302
Revises: 
Create Date: 2019-05-15 11:48:45.805342

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '3b63c8e4c302'
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('test',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('test1', sa.String(length=50), nullable=False),
    sa.Column('test2', sa.String(length=120), nullable=False),
    sa.Column('create_at', sa.DateTime(), nullable=True),
    sa.Column('update_at', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('test2')
    )
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('test')
    # ### end Alembic commands ###

查看版本：alembic history
- <base> -> 3b63c8e4c302 (head), init db
数据库升级：alembic upgrade 3b63c8

Database changed
mysql> show tables;
+------------------+
| Tables_in_qmasdb |
+------------------+
| alembic_version  |
| test             |
+------------------+
2 rows in set (0.00 sec)

mysql> desc test;
+-----------+--------------+------+-----+---------+----------------+
| Field     | Type         | Null | Key | Default | Extra          |
+-----------+--------------+------+-----+---------+----------------+
| id        | int(11)      | NO   | PRI | NULL    | auto_increment |
| test1     | varchar(50)  | NO   |     | NULL    |                |
| test2     | varchar(120) | NO   | UNI | NULL    |                |
| create_at | datetime     | YES  |     | NULL    |                |
| update_at | datetime     | YES  |     | NULL    |                |
+-----------+--------------+------+-----+---------+----------------+
5 rows in set (0.00 sec)

增加一个hahahahahahahat字段：

class Test(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    test1 = db.Column(db.String(50), unique=False, nullable=False)
    test2 = db.Column(db.String(120), unique=True, nullable=False)
    hahahahahahahat = db.Column(db.String(120), unique=True, nullable=False) # ╮(╯▽╰)╭
    create_at = db.Column(db.DateTime, unique=False, nullable=True, default=datetime.utcnow)
    update_at = db.Column(db.DateTime, unique=False, nullable=True, default=datetime.utcnow)

提交更新：alembic revision --autogenerate -m "add hahahahahahahat."

INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added column 'test.hahahahahahahat'
INFO  [alembic.autogenerate.compare] Detected added unique constraint 'None' on '['hahahahahahahat']'
  Generating /.../project_name/alembic/versions/290c92587e1a_add_hahahahahahahat.py ... done

让我们看看./alembic/versions/290c92587e1a_add_hahahahahahahat.py这个文件。

"""add hahahahahahahat.

Revision ID: 290c92587e1a
Revises: 3b63c8e4c302
Create Date: 2019-05-15 11:53:28.251516

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '290c92587e1a'
down_revision = '3b63c8e4c302'
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('test', sa.Column('hahahahahahahat', sa.String(length=120), nullable=False))
    op.create_unique_constraint(None, 'test', ['hahahahahahahat'])
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint(None, 'test', type_='unique')
    op.drop_column('test', 'hahahahahahahat')
    # ### end Alembic commands ###

看下提交记录：alembic history

3b63c8e4c302 -> 290c92587e1a (head), add hahahahahahahat.
<base> -> 3b63c8e4c302, init db

升级数据库：alembic upgrade 290c92

INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade 3b63c8e4c302 -> 290c92587e1a, add hahahahahahahat.

看看数据库

mysql> desc test;
+-----------------+--------------+------+-----+---------+----------------+
| Field           | Type         | Null | Key | Default | Extra          |
+-----------------+--------------+------+-----+---------+----------------+
| id              | int(11)      | NO   | PRI | NULL    | auto_increment |
| test1           | varchar(50)  | NO   |     | NULL    |                |
| test2           | varchar(120) | NO   | UNI | NULL    |                |
| create_at       | datetime     | YES  |     | NULL    |                |
| update_at       | datetime     | YES  |     | NULL    |                |
| hahahahahahahat | varchar(120) | NO   | UNI | NULL    |                | # 参数添加进来了╮(╯▽╰)╭
+-----------------+--------------+------+-----+---------+----------------+
6 rows in set (0.00 sec)

使用Alembic的好处就是可以控制数据库的版本，增加字段不影响数据。
先试试版本更新的。尝试增加wtfwtfwtfwtfwtf字段，带unique关键字……
- 提交：alembic revision --autogenerate -m "add wtfwtfwtfwtf."
- 更新数据库到最新版本：alembic upgrade head
- 降级到原版本：alembic downgrade -1

class Test(db.Model):
    ...
    wtfwtfwtfwtfwtf = db.Column(db.String(120), unique=True, nullable=False)
    ...

直接报错了……TypeError: object of type 'NoneType' has no len()
试试添加一个不带unique关键字的字段wawawawawawawawa

class Test(db.Model):
    ...
    wawawawawawawawa = db.Column(db.String(120), nullable=False)
    ...

更新：alembic revision --autogenerate -m "add wawawawawawa."
升级：alembic upgrade head

mysql> select test.id, test.wawawawawawawawa from test;
+----+------------------+
| id | wawawawawawawawa |
+----+------------------+
|  1 |                  |
|  2 | wawawawawawawawa |
+----+------------------+
2 rows in set (0.00 sec)

降级一波：alembic downgrade -1

INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running downgrade ad3ed6b3652c -> de464ecfaa3c, add wawawawawawa.

# 数据库
mysql> select test.id, test.wawawawawawawawa from test;
ERROR 1054 (42S22): Unknown column 'test.wawawawawawawawa' in 'field list'

重新升级回去：alembic upgrade head

mysql> select test.id, test.wawawawawawawawa from test;
+----+------------------+
| id | wawawawawawawawa |
+----+------------------+
|  1 |                  |
|  2 |                  |
+----+------------------+
2 rows in set (0.00 sec)

Alembic升级不会影响数据，但是降级一定丢数据。
文档中写了一句话：We review and modify these by hand as needed, then proceed normally.安全起见，每次提交之后最好review下自动生成的脚本(./alembic/versions/....py)。

其他操作

更新数据库：alembic upgrade 版本号
更新到最新版：alembic upgrade head
升两级：alembic upgrade +2
降级到最初版：alembic downgrade base
降级数据库：alembic downgrade 版本号
降两级：alembic downgrade -2
查看当前版本：alembic current
查看历史版本：
- alembic history --verbose
- alembic history -r版本号:版本号
  - alembic history -r-1:current：上一个版本到当前版本
  - alembic history -r8b6caa:6f406f：8b6caa版本（包含）到6f406f版本（包含）
  - alembic history -r8b6caa:：8b6caa版本（包含）到当前版本
离线更新(生成SQL)：alembic upgrade 版本号 --sql > migration.sql
从特定起始版本生成SQL：alembic upgrade 版本一:版本二 --sql > migration.sql
查询当前数据库版本号：
- 进入配置的数据库可以看到一个表alembic_version，其中的version_num关键字就是版本号。
清除所有版本：
- 需要将versions下的所有文件删除，并删除数据库表alembic_version。

查看支持的模板

alembic list_templates

Available templates:

multidb - Rudimentary multi-database configuration.
generic - Generic single-database configuration.
pylons - Configuration that reads from a Pylons project environment.

Templates are used via the 'init' command, e.g.:

  alembic init --template generic ./scripts

附录

级联删除

来源于：https://www.v2ex.com/t/507065

from flask import Flask
from flask_sqlalchemy import Model, SQLAlchemy
from sqlalchemy import Column, Integer, DateTime, func, String, ForeignKey
from sqlalchemy.orm import relationship, backref


class BaseModel(Model):
    id = Column(Integer, primary_key=True)
    created_at = Column(DateTime(True), default=func.now(), nullable=False)
    updated_at = Column(DateTime(True), default=func.now(), onupdate=func.now(), nullable=False)

    @classmethod
    def create(cls, **kw):
        session = db.session
        if 'id' in kw:
            obj = session.query(cls).get(kw['id'])
            if obj:
                return obj
        obj = cls(**kw)
        session.add(obj)
        session.commit()
        return obj

    def to_dict(self):
        columns = self.__table__.columns.keys()
        return {key: getattr(self, key) for key in columns}


app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqldb://root:iqiyi123@localhost:3306/test'

db = SQLAlchemy(app, model_class=BaseModel)


######################################################################################################

class Parent(db.Model):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    name = Column(String(20))


class Child(db.Model):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    name = Column(String(20))

    # 无法删除
    parent_id = Column(Integer, ForeignKey('parent.id'))
    parent = relationship("Parent")

    # 父级，子级一起删除
    # parent_id = Column(Integer, ForeignKey('parent.id', ondelete="CASCADE"))
    # parent = relationship("Parent", backref=backref("child", passive_deletes=True))

    # 父级删除，子级不删除，外键更新为null
    # parent_id = Column(Integer, ForeignKey('parent.id', ondelete="CASCADE"))
    # parent = relationship("Parent", backref=backref("child"))


db.create_all()
Parent.create(name='ZhangTian')
Parent.create(name='LiTian')
Child.create(name='ZhangDi', parent_id=1)
Child.create(name='LiDi', parent_id=2)

parent = db.session.query(Parent).first()
db.session.delete(parent)
db.session.commit()