ScoDoc/scodoc.py

# -*- coding: UTF-8 -*


"""Application Flask: ScoDoc


"""


from __future__ import print_function

import os
from pprint import pprint as pp
import sys

import click
import flask
from flask.cli import with_appcontext
from app import create_app, cli, db
from app import initialize_scodoc_database
from app import clear_scodoc_cache

from app.auth.models import User, Role, UserRole
from app import models
from app.models import ScoPreference

from app.views import notes, scolar, absences
import tools

from config import DevConfig

app = create_app(DevConfig)
cli.register(app)


@app.shell_context_processor
def make_shell_context():
    from app.scodoc import notesdb as ndb
    from app.scodoc import sco_utils as scu
    from flask_login import login_user, logout_user, current_user
    import app as mapp  # le package app

    return {
        "db": db,
        "User": User,
        "Role": Role,
        "UserRole": UserRole,
        "notes": notes,
        "scolar": scolar,
        "ndb": ndb,
        "scu": scu,
        "pp": pp,
        "flask": flask,
        "current_app": flask.current_app,
        "current_user": current_user,
        "login_user": login_user,
        "logout_user": logout_user,
        "ctx": app.test_request_context(),
        "models": models,
        "mapp": mapp,
    }


# ctx.push()
# login_user(admin)


@app.cli.command()
def db_init():  # db-init
    """Initialize the database.
    Starts from an existing database and create all
    the necessary SQL tables and functions.
    """
    initialize_scodoc_database()


@app.cli.command()
def user_db_clear():
    """Erase all users and roles from the database !"""
    click.echo("Erasing the users database !")
    _clear_users_db()


def _clear_users_db():
    """Erase (drop) all tables of users database !"""
    click.confirm(
        "This will erase all users and roles.\nAre you sure you want to continue?",
        abort=True,
    )
    db.reflect()
    try:
        db.session.query(UserRole).delete()
        db.session.query(User).delete()
        db.session.query(User).delete()
        db.session.commit()
    except:
        db.session.rollback()
        raise


@app.cli.command()
@click.argument("username")
@click.argument("role")
@click.argument("dept")
@click.option("-n", "--nom", "nom")
@click.option("-p", "--prenom", "prenom")
def user_create(username, role, dept, nom=None, prenom=None):  # user-create
    "Create a new user"
    r = Role.get_named_role(role)
    if not r:
        sys.stderr.write("user_create: role {r} does not exists\n".format(r=role))
        return 1
    u = User.query.filter_by(user_name=username).first()
    if u:
        sys.stderr.write("user_create: user {u} already exists\n".format(u=u))
        return 2
    if dept == "@all":
        dept = None
    u = User(user_name=username, dept=dept, nom=nom, prenom=prenom)
    u.add_role(r, dept)
    db.session.add(u)
    db.session.commit()
    click.echo(
        "created user, login: {u.user_name}, with role {r} in dept. {dept}".format(
            u=u, r=r, dept=dept
        )
    )


@app.cli.command()
@click.argument("username")
@click.password_option()
def user_password(username, password=None):  # user-password
    "Set (or change) user's password"
    if not password:
        sys.stderr.write("user_password: missing password")
        return 1
    u = User.query.filter_by(user_name=username).first()
    if not u:
        sys.stderr.write("user_password: user {} does not exists".format(username))
        return 1

    u.set_password(password)
    db.session.add(u)
    db.session.commit()
    click.echo("changed password for user {}".format(u))


@app.cli.command()
@click.argument("dept")
def delete_dept(dept):  # delete-dept
    """Delete existing departement"""
    from app.scodoc import notesdb as ndb
    from app.scodoc import sco_dept

    click.confirm(
        f"""Attention: Cela va effacer toutes les données du département {dept}
        (étudiants, notes, formations, etc)
        Voulez-vous vraiment continuer ?
        """,
        abort=True,
    )
    db.reflect()
    ndb.open_db_connection()
    d = models.Departement.query.filter_by(acronym=dept).first()
    if d is None:
        sys.stderr.write(f"Erreur: le departement {dept} n'existe pas !")
        return 2
    sco_dept.delete_dept(d.id)
    db.session.commit()
    return 0


@app.cli.command()
@click.argument("dept")
def create_dept(dept):  # create-dept
    "Create new departement"
    d = models.Departement(acronym=dept)
    p1 = ScoPreference(name="DeptName", value=dept, departement=d)
    db.session.add(p1)
    db.session.add(d)
    db.session.commit()
    return 0


@app.cli.command()
@click.argument("filename")
@with_appcontext
def test_interactive(filename=None):
    "Run interactive test"
    import flask_login
    from app import decorators

    click.echo("Executing {}".format(filename))
    with app.test_request_context(""):
        u = User.query.first()
        flask_login.login_user(u)
        REQUEST = decorators.ZRequest()
        exec(open(filename).read())

    click.echo("Done.")


@app.cli.command()
@with_appcontext
def import_scodoc7_users():  # import-scodoc7-users
    """Import used defined in ScoDoc7 postgresql database into ScoDoc8
    The old database SCOUSERS  must be alive and readable by the current user.
    This script is typically run as unix user "scodoc".
    The original SCOUSERS database is left unmodified.
    """
    messages = tools.import_scodoc7_user_db()
    click.echo("----")
    click.echo(f"import terminé: {len(messages)} warnings\n")
    click.echo("\n".join(messages) + "\n")


@app.cli.command()
@click.argument("dept")
@with_appcontext
def import_scodoc7_dept(dept):  # import-scodoc7-dept
    """Import département ScoDoc7"""
    tools.import_scodoc7_dept(dept)


@app.cli.command()
@with_appcontext
def clear_cache():  # clear-cache
    """Clear ScoDoc cache
    This cache (currently Redis) is persistent between invocation
    and it may be necessary to clear it during developement or tests.
    """
    clear_scodoc_cache()
    click.echo("Redis caches flushed.")