terminate.jpg

首页

技术分享

创建带数据库和迁移脚本的项目

首先安装必要的依赖包

Flask==2.0.1
Flask-SQLAlchemy==2.5.1
Flask-Migrate==3.1.0
# 以上是需要手动安装的包,安装后会自动安装依赖包,完整的包列表如下


alembic==1.7.1
click==8.0.1
colorama==0.4.4
Flask==2.0.1
Flask-Migrate==3.1.0
Flask-SQLAlchemy==2.5.1
greenlet==1.1.1
itsdangerous==2.0.1
Jinja2==3.0.1
Mako==1.1.5
MarkupSafe==2.0.1
SQLAlchemy==1.4.23
Werkzeug==2.0.1

项目根目录添加如下脚本

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///./test.db'
db =SQLAlchemy(app)
migrate = Migrate(app, db)

class TestModel(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)

def __repr__(self):
return '<User %r>' % self.username


@app.route("/")
def hello_world():
return "<p>Hello, World!</p>"

然后执行命令

flask db init
flask db migrate -m "initial a project database struct"
flask db upgrade

用数据库客户端打开test.db 文件或自定义的数据库位置,可以看见数据库表已经被创建成功了

这个时候还有个问题,migrate出来的迁移文件不会导入自定义的类型所在的包,需要手动添加上去

"""empty message

Revision ID: c59c30a6314b
Revises:
Create Date: 2021-09-06 17:20:30.250378

"""
from alembic import op
import sqlalchemy as sa
from custom_fields import myfield # 这一句不会自动导入

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


def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('test_model',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('username', sa.String(length=80), nullable=False),
sa.Column('email', sa.String(length=120), nullable=False),
sa.Column('custom_field', myfield.MyField(), nullable=True),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('email'),
sa.UniqueConstraint('username')
)
# ### end Alembic commands ###

这是可以在flask db init的时候创建出来的migrations/env.py里手动修改一下

新增一个render_item 函数,然后把这个函数传递给context.configure字典

from __future__ import with_statement

import logging
from logging.config import fileConfig

from flask import current_app

from alembic import context

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
fileConfig(config.config_file_name)
logger = logging.getLogger('alembic.env')

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
config.set_main_option(
'sqlalchemy.url',
str(current_app.extensions['migrate'].db.get_engine().url).replace(
'%', '%%'))
target_metadata = current_app.extensions['migrate'].db.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.

##__此函数为新增函数___________##########################################
from custom_fields.myfield import MyField
def render_item(type_, obj, autogen_context):
"""Apply custom rendering for selected items."""

if type_ == 'type' and isinstance(obj, MyField):
autogen_context.imports.add("from custom_fields import myfield")
return "myfield.%r" % obj

# default rendering for other objects
return False
#####################################################################

def run_migrations_offline():
"""Run migrations in 'offline' mode.

This configures the context with just a URL
and not an Engine, though an Engine is acceptable
here as well. By skipping the Engine creation
we don't even need a DBAPI to be available.

Calls to context.execute() here emit the given string to the
script output.

"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url, target_metadata=target_metadata, literal_binds=True,
render_item=render_item, # 此项为新增项
)

with context.begin_transaction():
context.run_migrations()


def run_migrations_online():
"""Run migrations in 'online' mode.

In this scenario we need to create an Engine
and associate a connection with the context.

"""

# this callback is used to prevent an auto-migration from being generated
# when there are no changes to the schema
# reference: http://alembic.zzzcomputing.com/en/latest/cookbook.html
def process_revision_directives(context, revision, directives):
if getattr(config.cmd_opts, 'autogenerate', False):
script = directives[0]
if script.upgrade_ops.is_empty():
directives[:] = []
logger.info('No changes in schema detected.')

connectable = current_app.extensions['migrate'].db.get_engine()

with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
process_revision_directives=process_revision_directives,
render_item=render_item, # 此项为新增项
**current_app.extensions['migrate'].configure_args
)

with context.begin_transaction():
context.run_migrations()


if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

但是,这样依然不方便

Comments:

captcha