vowel/run.py

#!/usr/bin/env python3
# coding=utf-8
#
# run.py: Run vowel
# Copyright (C) 2015, 2017 Sam Black <samwwwblack@lapwing.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
import eventlet
eventlet.monkey_patch()  # noqa

import sys

import click

from flask_migrate import upgrade as db_upgrade
from sqlalchemy_utils.functions import create_database
from sqlalchemy_utils.functions import database_exists

from vowel.models import db
from vowel.utils.user import AddUserError
from vowel.utils.user import add_user
from vowel.utils.user import check_email_validity
from vowel.utils.user import create_default_roles
from vowel.web import create_app


app = create_app()


def _check_db_exists():
    """
    Check if the database exists.

    :return: True if DB exists, False otherwise
    :rtype: bool
    """
    db_url = app.config["SQLALCHEMY_DATABASE_URI"]
    return database_exists(db_url)


def _check_db_tables_exist():
    """
    Check if the database exists and has tables.

    :return: True if the DB exists and has tables, False otherwise
    :rtype: bool
    """
    return _check_db_exists() and db.engine.dialect.has_table(db.engine,
                                                              "user")


def _init_db():
    """
    Create the database, tables and required starting data.
    """
    db_url = app.config["SQLALCHEMY_DATABASE_URI"]

    if db_url.startswith("postgres://"):
        click.echo(
            ("'SQLALCHEMY_DATABASE_URI' must start with 'postgresql://', "
             "not 'postgres://'; please update your instance config."))
        sys.exit(1)

    if not _check_db_exists():
        create_database(db_url)

    if _check_db_tables_exist():
        click.echo("Database already exists.")
        click.echo("Please check your installation.")
        sys.exit(1)

    db_upgrade()
    create_default_roles()
    db.session.commit()


@app.cli.command()
def init_db():
    """
    Create the database and populate the required data.
    """
    _init_db()


@app.cli.command()
def create_user():
    """
    Create a new user.
    """
    if not _check_db_tables_exist():
        click.echo("Database does not exist yet.")
        click.echo("Please check your installation, "
                   "or run 'FLASK_APP=run.py flask first_run'")
        sys.exit(1)

    email = None
    while not email:
        email_tmp = click.prompt("Email")

        try:
            check_email_validity(email_tmp)
        except AddUserError:
            click.echo("Email is invalid")
        else:
            email = email_tmp

    password = None
    while not password:
        password = click.prompt("Password", hide_input=True,
                                confirmation_prompt=True)

    admin = click.confirm("Administrator?")

    roles = ("Admin",) if admin else ()

    add_user(email, password, roles)

    click.echo("User added")


@app.cli.command()
def first_run():
    """
    Init the database and
    query the user for an administrator email/password.
    """
    _init_db()

    email = None
    while not email:
        email_tmp = click.prompt("Administrator email")

        try:
            check_email_validity(email_tmp)
        except AddUserError:
            click.echo("Email is invalid")
        else:
            email = email_tmp

    password = None
    while not password:
        password = click.prompt("Administrator Password",
                                hide_input=True, confirmation_prompt=True)

    add_user(email, password, ("Admin",), True)

    click.echo("Setup complete")