CLI support

flask_sqlalchemy/__init__.py

@@ -780,6 +780,7 @@ def init_app(self, app):
             )
 
         app.config.setdefault('SQLALCHEMY_DATABASE_URI', 'sqlite:///:memory:')
+        app.config.setdefault('SQLALCHEMY_CLI_DB_NAME', 'db')
         app.config.setdefault('SQLALCHEMY_BINDS', None)
         app.config.setdefault('SQLALCHEMY_NATIVE_UNICODE', None)
         app.config.setdefault('SQLALCHEMY_ECHO', False)
@@ -802,6 +803,12 @@ def init_app(self, app):
 
         app.extensions['sqlalchemy'] = _SQLAlchemyState(self)
 
+        if hasattr(app, 'cli'):
+            from .cli import db
+            cli_name = app.config['SQLALCHEMY_CLI_DB_NAME']
+            if cli_name:
+                app.cli.add_command(db, cli_name)
+
         @app.teardown_appcontext
         def shutdown_session(response_or_exc):
             if app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN']:

flask_sqlalchemy/cli.py

@@ -0,0 +1,49 @@
+# -*- coding: utf-8 -*-
+"""
+    flask_sqlalchemy.cli
+    ~~~~~~~~~~~~~~~~~~~~
+
+    Command Line Interface for managing the database.
+
+    :copyright: (c) 2017 by David Baumgold.
+    :license: MIT, see LICENSE for more details.
+"""
+
+from __future__ import absolute_import
+
+from functools import wraps
+
+import click
+from flask import current_app
+from werkzeug.local import LocalProxy
+
+try:
+    from flask.cli import with_appcontext
+except ImportError:
+    from flask_cli import with_appcontext
+
+state = LocalProxy(lambda: current_app.extensions['sqlalchemy'])
+
+
+def commit(fn):
+    """Decorator to commit changes after the command is run."""
+    @wraps(fn)
+    def wrapper(*args, **kwargs):
+        fn(*args, **kwargs)
+        state.db.session.commit()
+    return wrapper
+
+
+@click.group()
+def db():
+    """Manages the database."""
+
+
+@db.command('create')
+@click.option('--bind', default='__all__')
+@with_appcontext
+@commit
+def db_create(bind):
+    """Creates database tables."""
+    state.db.create_all(bind=bind)
+    click.secho('Database created successfully.', fg='green')

tests/conftest.py

@@ -3,6 +3,7 @@
 import flask
 import pytest
 
+import sqlalchemy
 import flask_sqlalchemy as fsa
 
 
@@ -12,6 +13,10 @@ def app(request):
     app.testing = True
     app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
     app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
+    app.config['SQLALCHEMY_BINDS'] = {
+        'users': 'mysqldb://localhost/users',
+        'geo':   'postgres://localhost/geo',
+    }
     return app
 
 
@@ -38,3 +43,48 @@ def __init__(self, title, text):
     db.create_all()
     yield Todo
     db.drop_all()
+
+
+@pytest.fixture
+def clirunner():
+    from click.testing import CliRunner
+    return CliRunner()
+
+
+@pytest.fixture
+def script_info(app, db):
+    try:
+        from flask.cli import ScriptInfo
+    except ImportError:
+        from flask_cli import ScriptInfo
+
+    return ScriptInfo(create_app=lambda x: app)
+
+
+@pytest.fixture(autouse=True)
+def mock_engines(mocker):
+    """Mock all SQLAlchemy engines, except SQLite
+    (which requires no dependencies)"""
+    real_create_engine = sqlalchemy.create_engine
+    real_EngineDebuggingSignalEvents = fsa._EngineDebuggingSignalEvents
+
+    def mock_create_engine(info, **options):
+        # sqlite has no dependencies, so we won't mock it
+        if info.drivername == 'sqlite':
+            return real_create_engine(info, **options)
+        # every other engine has dependencies, so we'll mock them
+        return mocker.Mock(name="{info.drivername} engine".format(info=info))
+
+    def mock_debugging_signals(engine, import_name):
+        if isinstance(engine, mocker.Mock):
+            return mocker.Mock(name=engine._mock_name + " debugging signals")
+        return real_EngineDebuggingSignalEvents(engine, import_name)
+
+    mocker.patch(
+        'flask_sqlalchemy.sqlalchemy.create_engine',
+        mock_create_engine,
+    )
+    mocker.patch(
+        'flask_sqlalchemy._EngineDebuggingSignalEvents',
+        mock_debugging_signals,
+    )

tests/test_cli.py

@@ -0,0 +1,16 @@
+import pytest
+pytest.importorskip("click")
+
+from flask_sqlalchemy.cli import db_create
+
+
+def test_cli_create(clirunner, db, script_info, mocker):
+    mock_execute = mocker.patch.object(
+        db,
+        "_execute_for_all_tables",
+    )
+
+    result = clirunner.invoke(db_create, obj=script_info)
+    assert result.exit_code == 0
+    mock_execute.assert_called_once_with(None, '__all__', 'create_all')
+    assert "Database created successfully." in result.output

tox.ini

@@ -8,6 +8,7 @@ envlist =
 [testenv]
 deps =
     pytest>=3
+    pytest-mock
     coverage
     blinker