SqlAlchemy + Alembic Migrate

SqlAlchemy+sqlite3+Alembic 数据库迁移踩坑…

NotImplementedError: No support for ALTER of constraints in SQLite dialect.
Constraint must have a name

迁移需求：

1. 因为需求不断变化，所以数据库的字段甚至是table变化会比较大，自动迁移非常重要。
2. 数据库是分布式的，每个user都有自己的db
3. 丢弃迁移生成的文件(对我来说不重要)，只需要保证数据库是最新的版本

对于有编程需求的来说，官网文档太折磨人了。。。
我只想设置几个变量来替换默认的参数，但看了几遍文档，似乎并没找到。
唯一简单点的参数就只有script_location和sqlalchemy.url了，但这也仅仅针对于ini文件。
想改环境变量中的script_location，url，我也只能改已经生成的env.py文件了，最好的方式应该是设置好这些环境变量，在生成脚本的时候直接替换默认的环境变量，但由于时间紧迫，只能先用笨办法了，后面有机会再深究…

详细步骤(Function)：

1. 先实例化Config，把script_location和sqlalchemy.url传进去 --> setup_config
2. 执行init生成ini文件和env.py以及其他相关文件(在init之前会先清空目录，这里是因为我有特殊需求，一旦version信息被清除，数据库就无法退回了，谨慎操作！！！) --> init
3. 修改env.py 文件
    a. sqlalchemy.url
    b. metadata
    c. 修改alembic.context.configure的参数 render_as_batch=True
    d. *.ini 文件也是要改sqlalchemy.url，不知道为什么config设置了，但是.ini文件中还是默认的url
4. clear alembic version table. 由于迁移相关的文件都是在user目录下面的，所以可能会被删。这些文件里面有个version，每次迁移都会生成一个新的，而且会在数据库建一个alembic_version的table一旦两边的version版本对不上或者本地的version被删除，再迁移的时候就会报错，由于我不需要这些东西，只需要保证数据库是最新的，所以这里我直接暴力删除数据库中上一次alembic_version的数据。注意：这样做风险比较大，有restore需求的不能这样做。-->clear_alembic_version_table
5. revision --> revision
6. upgrade --> upgrade

踩坑一： NotImplementedError: No support for ALTER of constraints in SQLite dialect.

batch_mode
这个是sqlite3的坑，普通模式下，删除列之后再迁移就会报这个错，加外键也会。
解决办法：使用批处理模式 batch_mode
在环境变量的configure中设置参数render_as_batch=True，由于没找到怎么设置默认的环境变量并使之生效，所以这里简单粗暴的去改了env.py，[详细步骤]中的3.c。

踩坑二： Constraint must have a name
外网解决办法
当我把坑一的问题解决之后，我以为已经大功告成了，结果又出现另一个报错…

百度了一圈，倒是有解决办法，但都是一模一样，抄的外网论坛的方法，而且是flask-migrate。
但还是有点启发的，目的都一样就是给Base.metadata 设置 naming_convention具体代码在下面models的部分
在models中设置naming_convention之后，env.py import Base之后就生效了

migrate完整代码：

import re
import os
import sys
import shutil
from sqlalchemy import create_engine

from alembic import command
from alembic.config import Config

models_path = os.path.abspath(os.path.dirname(__file__))
sys.path.append(models_path)

class Migration:

    def __init__(self, db_path, script_location):
        self.db_path = db_path
        self.script_location = script_location
        self.alem_ini = f"{self.script_location.rstrip('alembic')}/alembic.ini"
        self.alembic_cfg = Config(self.alem_ini)

    def setup_config(self):
        self.alembic_cfg.set_main_option("script_location", self.script_location)
        self.alembic_cfg.set_main_option(
            "sqlalchemy.url", f"sqlite:///{self.db_path}?check_same_thread=False")

    def init(self):
        if os.path.exists(self.script_location):
            shutil.rmtree(self.script_location)
        command.init(self.alembic_cfg, self.script_location)

    def modify_env_file(self):
        alem_env = f"{self.script_location}/env.py"
        alem_url = f"sqlalchemy.url = sqlite:///{self.db_path}?check_same_thread=False\n"
        alem_meta = f"import sys\nsys.path.append('{models_path}')\nfrom models import Base\ntarget_metadata=Base.metadata\n"
        env_para = f"render_as_batch=True, compare_type=True,"
        env_fix_bug = ""
        with open(self.alem_ini)as ini_file:
            ini_lines = ini_file.readlines()
        mdf_ini_lines = [alem_url if i.startswith("sqlalchemy.url") else i for i in ini_lines]
        with open(self.alem_ini, "w") as ini_file:
            ini_file.write("".join(mdf_ini_lines))

        with open(alem_env) as env_file:
            env_lines = env_file.readlines()
        mdf_env_lines = [alem_meta if i.startswith("target_metadata") else i for i in env_lines]
        new_env_lines = []
        for line in mdf_env_lines:
            new_env_lines.append(line)
            if line.strip() == "context.configure(":
                space = re.search(r"\s+", line).group()
                new_env_lines.append(f"{space}{env_para}\n")
        with open(alem_env, "w") as env_file:
            env_file.write("".join(new_env_lines))

    def revision(self):

        command.revision(self.alembic_cfg, autogenerate=True)

    def upgrade(self):
        command.upgrade(self.alembic_cfg, 'head')

    def clear_alembic_version_table(self):
        engine = create_engine(f"sqlite:///{self.db_path}?check_same_thread=False")
        with engine.begin() as connection:
            result = connection.execute("SELECT COUNT(*) FROM sqlite_master where type ='table' and name = 'alembic_version'")
            if result.first()[0]:
                connection.execute("delete from alembic_version")

def main(db_path, script_location):
    migrate = Migration(db_path, script_location)
    migrate.setup_config()
    migrate.init()
    migrate.modify_env_file()
    migrate.clear_alembic_version_table()
    migrate.revision()
    migrate.upgrade()

models：

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
naming_convention = {
    "ix": 'ix_%(column_0_label)s',
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(column_0_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s"}

Base.metadata.naming_convention=naming_convention

Original: https://blog.csdn.net/w392377436/article/details/127498212
Author: self-discipline.
Title: SqlAlchemy + Alembic Migrate