如何以编程方式运行/调用烧瓶 cli 命令？

Question

我正在使用 python 3 和烧瓶，使用烧瓶迁移（使用 alembic）来处理我的 SQL 迁移。当我运行本地集成测试时，我想每次都重建数据库，以便我可以针对我正在测试的每个 api 调用的干净数据库运行我的 API 调用（是的，我可以使用 sqlite，但我想检查约束是否正确）。

我可以在命令行上轻松地执行以下操作：

mysql -uroot -e 'drop database DBNAME; create database DBNAME;'
FLASK_APP=flask_app.py flask db upgrade

但我宁愿在 python 代码中运行它，原因有两个：

1. 我不想担心安装在 CI 机器上的 mysql 客户端（最终）将运行此代码（它们应该只需要 python mysql 包）。
2. 我想操纵烧瓶设置来强制数据库名称以避免意外（因此它需要与调用它的脚本在同一线程/内存空间中运行）。

app 对象（使用app = Flask(__name__) 创建）有一个cli 属性，但它需要一个上下文对象，而且感觉我使用的工具不正确。我期待app.cli.invoke('db', 'upgrade') 或类似的...

关于如何在没有子 cli 进程的情况下从代码调用烧瓶命令的任何建议？

Answer 1

这不是很好，但最后我避免直接​​使用烧瓶命令，这似乎可以满足我的需要：

from my.app import app, db, initialize_app
from flask_migrate import Migrate
from alembic import command
from my.settings import settings
from sqlalchemy_utils.functions import drop_database, create_database, database_exists

test_db_name = 'test_db'
db_url = f'mysql+pymysql://mysqluser@127.0.0.1/{test_db_name}'
settings.SQLALCHEMY_DATABASE_URI = db_url


def reset():
    if database_exists(db_url):
        drop_database(db_url)
    create_database(db_url)
    initialize_app(app) # sets flask config SQLALCHEMY_DATABASE_URI to include test_db
    with app.app_context():
        config = Migrate(app, db).get_config()
        command.upgrade(config, 'head')

Answer 2

我使用以下模式（见下文）。在https://flask.palletsprojects.com/en/1.1.x/cli/?highlight=click#application-context

可以看到另一种方法

# file: commands.py
import click
from click import pass_context
from flask.cli import AppGroup, with_appcontext
from flask import current_app
from flask_migrate import Migrate
from alembic import command

from extensions import flask_db as db

db_cli = AppGroup('db', help='Various database management commands.')

@db_cli.command('init')
def db_init():
    """Initialize the database."""
    db.create_all()
    click.echo("Create all tables.")


@db_cli.command('drop')
def db_drop():
    """Drop the database."""
    db.engine.execute("SET FOREIGN_KEY_CHECKS=0;")
    db.drop_all()
    db.engine.execute("SET FOREIGN_KEY_CHECKS=1;")
    click.echo("Drop all tables.")

@db_cli.command('migrate')
def db_migrate():
    "Migrate with alembic."

    config = Migrate(current_app, db).get_config()
    command.upgrade(config, 'head')


@db_cli.command('db_upgrade')
@pass_context
def db_upgrade(ctx):
    """Alias for 'db reset'."""
    db_drop.invoke(ctx)
    db_init.invoke(ctx)
    db_migrate.invoke(ctx)

# file: extensions.py
# Keep your extenstions separate to allow importing without import loops.

from flask_sqlalchemy import SQLAlchemy

flask_db = SQLAlchemy()

# file: app.py (app/__init__.py) wherever your app is built
from extensions import flask_db

app = Flask(__name__)

flask_db.init_app(app)  # I'm not sure if the order matters here.
app.cli.add_command(db_cli)

# file: wsgi.py (top level file)
# This file lets you run 'flask' commands (e.g. flask routes)

# noinspection PyUnresolvedReferences
from app import app as application  # noqa

# file layout
- /
  - app/  (or app.py)
    - __init__.py  (optional)
  - commands.py
  - extensions.py
  - wsgi.py

用法：flask db upgrade

Answer 3

有什么理由不能简单地包装具有 CLI 的所有逻辑的函数？

类似

# cli.py

def db_init(cli=False):
    db.create_all()
    if cli:
        print("Created all tables")

@app.cli.command('db-init')
def _db_init(): # I'm just a wrapper function
    db_init(cli=True) 

如果你这样做，你应该没有理由不能在以后简单地这样做：

# some_module.py

from .cli import db_init

db_init()