AdvLabDB/advlabdb/actions.py

import subprocess
from datetime import datetime

from flask import flash
from flask_login import current_user
from sqlalchemy import select

from . import data_dir
from .models import Assistant, Semester, User, db


def update_final_experiment_and_part_marks():
    for semesterExperiment in current_user.active_semester.semester_experiments:
        semesterExperiment.updateFinalExperimentAndPartMarks()

    flash("Manually updated all final experiment and part marks.", "success")


def backup(backup_prefix):
    db_path = data_dir / "db/advlabdb.sqlite"

    db_bk_dir = db_path.parent / "backups"
    db_bk_dir.mkdir(exist_ok=True)

    now = datetime.now().strftime("%d_%m_%Y_%H_%M_%S")
    dest = db_bk_dir / f"{backup_prefix}_{now}.sqlite"

    status = subprocess.run(["sqlite3", db_path, f".backup {dest}"], check=False).returncode
    if status == 0:
        flash(f"Created a database backup at the path {dest}", "success")
    else:
        flash("Failed to create a database backup! Make sure that `sqlite3` is installed on your system!", "danger")


def manual_backup():
    backup("manual")


def deactivate_assistants():
    user_ids_to_deactivate = db.session.scalars(
        select(Assistant.user_id)
        .join(User)
        .where(User.active is True)
        .except_(
            select(Assistant.user_id).join(Assistant.semester_experiments).join(Semester).where(Semester.done is False)
        )
    )

    no_users_deactivated = True

    try:
        for user_id in user_ids_to_deactivate:
            user = db.session.get(User, user_id)
            user.active = False

            flash(f"User {user} deactivated!", "warning")
            no_users_deactivated = False

        db.session.commit()
    except Exception as ex:
        flash(str(ex), "danger")

        db.session.rollback()
        no_users_deactivated = True

    if no_users_deactivated:
        flash("No users to deactivate.", "info")