如何以编程方式运行/调用Flask CLI命令？

Question

我正在使用Python 3和Flask，同时使用flask-migrate（它使用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. 我希望通过操纵Flask设置来强制数据库名称以避免意外情况（因此它需要在调用它的脚本的同一线程/内存空间中运行）。

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

有没有关于如何从代码中调用Flask命令而无需子cli进程的建议？

- Sam Adams

---

我认为这篇SO帖子可以帮到您。 - Kristjan Tärk

Answer 1

我使用以下模式（见下文）。另一种方法可以在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

- Marlen T. B.

Answer 2

虽然不是很完美，但最终我避免了直接使用flask命令，这似乎达到了我所需的效果：

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')

- Sam Adams

---

这种方法有一个缺点。通常情况下，您希望将数据库用户分配给应用程序，而不是拥有完整的管理员权限，而只有足够的权限来处理一个数据库。如果使用管理员用户进行应用程序，则存在安全风险，如果有人获得了您的数据库管理员凭据，则可以清除您的数据库服务器。标准做法是为应用程序使用受限用户，该用户仅具有对数据库的读写访问权限。在这些情况下，您可以使用 db.drop_all() 删除数据库中的所有表，而不删除数据库本身。 - Miguel Grinberg

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()

- JackofSpades