Merge pull request #18 from lendingblock/master

added create db command

openapi/__init__.py

@@ -1,3 +1,3 @@
 """Minimal OpenAPI asynchronous server application
 """
-__version__ = '0.1.1'
+__version__ = '0.1.2'

openapi/db/commands.py

@@ -1,5 +1,9 @@
+from copy import copy
+
 import click
 
+from sqlalchemy_utils import database_exists, drop_database, create_database
+
 from .migrations import Migration
 
 
@@ -71,3 +75,24 @@ def show(ctx, revision):
     """Show revision ID and creation date
     """
     return migration(ctx).show(revision)
+
+
+@db.command()
+@click.argument('dbname', nargs=1)
+@click.option('--force', default=False, is_flag=True,
+              help='Force removal of an existing database')
+@click.pass_context
+def create(ctx, dbname, force):
+    """Creates a new database
+    """
+    store = ctx.parent.parent.app['store']
+    url = copy(store.url)
+    url.database = dbname
+    store = str(url)
+    if database_exists(store):
+        if force:
+            drop_database(store)
+        else:
+            return click.echo(f'database {dbname} already available')
+    create_database(store)
+    click.echo(f'database {dbname} created')

tests/test_db.py

@@ -12,6 +12,15 @@ def test_db(cli):
     assert result.output.startswith('Usage: root db [OPTIONS]')
 
 
+def test_createdb(cli):
+    runner = CliRunner()
+    result = runner.invoke(cli.app['cli'], ['db', 'create', 'testing-aio-db'])
+    assert result.exit_code == 0
+    result = runner.invoke(
+        cli.app['cli'], ['db', 'create', 'testing-aio-db', '--force'])
+    assert result.exit_code == 0
+
+
 async def tests_get_list(cli):
     response = await cli.get('/tasks')
     data = await jsonBody(response)