python sqlalchemy-migrate 使用方法

时间:2022-07-15 16:01:31
1:下载相关模块
    pip install sqlalchemy
    pip install sqlalchemy-migrate
 
2:创建model (model.py),这里用来绑定数据库,创建sqlmodel,当然你也可以吧数据库设置和model分开。
#coding:utf-8
#made by dewei
from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base  # db 基类
from sqlalchemy import Column, Integer, String, DateTime  # 相应的列
from sqlalchemy.orm import sessionmaker, relationship, backref, scoped_session  # 执行的相关方法
from sqlalchemy.dialects.mysql import LONGTEXT
 
import os
"""
设置数据库参数
"""
MYSQL_DB = 'sqlalchemy_migratetest'
MYSQL_USER = 'root'
MYSQL_PASSWD = 'dewei'
MYSQL_HOST = 'localhost'
MYSQL_POST = 3306
 
basedir = os.path.abspath(os.path.dirname(__file__))
SQLALCHEMY_DATABASE_URI = "mysql://%s:%s@%s:%s/%s?charset=utf8" % (MYSQL_USER, MYSQL_PASSWD, MYSQL_HOST, MYSQL_POST, MYSQL_DB)
SQLALCHEMY_MIGRATE_REPO = os.path.join(basedir, 'db_repository')  # 设置数据库迁移保存的文件夹,用来sqlalchemymigrate
 
db = create_engine(SQLALCHEMY_DATABASE_URI, echo=True)
 
Base = declarative_base()
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=db))
 
Base.query = db_session.query_property()
 
def isenit_db():
    Base.metadata.create_all(bind=db)
 
class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    username = Column(String(100), unique=True, nullable=False)
    password = Column(String(200), nullable=False)
    email = Column(String(200), nullable=False)
 
    def __init__(self, username, password, email):
        self.username = username
        self.password = password
self.email = email
 
    def repr(self):
        return "" % (self.username, self.password)
 
3:生成 数据库并创建migrate迁移初始化 db_create.py
from model import SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO, db, Base
from migrate.versioning import api
import os.path
 
Base.metadata.create_all(bind=db)
 
if not os.path.exists(SQLALCHEMY_MIGRATE_REPO):
    api.create(SQLALCHEMY_MIGRATE_REPO, 'database repository')
    api.version_control(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO)
else:
    api.version_control(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO, api.version(SQLALCHEMY_MIGRATE_REPO))
 
当完成本py后 执行一次 仅一次就好, python db_create.py 这样数据库就创建好了 并且migrate也初始化完成,然后创建第四部,以后 添加数据库 就使用他就可以了
 
4 创建同步数据库py db_migrate.py
import imp
from migrate.versioning import api
from model import Base
from model import SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO
 
migration = SQLALCHEMY_MIGRATE_REPO + '/versions/d_migration.py' % (api.db_version(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO) + 1)
tmp_module = imp.new_module('old_model')
old_model = api.create_model(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO)
exec old_model in tmp_module.__dict__
script = api.make_update_script_for_model(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO, tmp_module.meta, Base.metadata)
open(migration, 'wt').write(script)
api.upgrade(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO)
print 'New migration saved as ' + migration
print 'Current database version: ' + str(api.db_version(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO))
 
完成后 试着修改一次数据库 比如给user添加个新的字段 然后执行 python db_migrate.py 去sql里看看 已经添加完成了!