models.py

# -*- coding: utf-8 -*-
"""
app.models
~~~~~~~~~~

This module implements the models used in the app.

:copyright: (c) 2017 European Spallation Source ERIC
:license: BSD 2-Clause, see LICENSE for more details.

"""
import ipaddress
import string
import qrcode
import itertools
import urllib.parse
import elasticsearch
import sqlalchemy as sa
from enum import Enum
from operator import attrgetter
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import validates
from sqlalchemy_continuum import make_versioned, version_class
from citext import CIText
from flask import current_app
from flask_login import UserMixin, current_user
from wtforms import ValidationError
from rq import Queue
from .extensions import db, login_manager, ldap_manager, cache
from .plugins import FlaskUserPlugin
from .validators import (
    ICS_ID_RE,
    HOST_NAME_RE,
    INTERFACE_NAME_RE,
    VLAN_NAME_RE,
    MAC_ADDRESS_RE,
    DEVICE_TYPE_RE,
)
from . import utils, search


make_versioned(plugins=[FlaskUserPlugin()])


# See http://docs.sqlalchemy.org/en/latest/core/compiler.html#utc-timestamp-function
class utcnow(sa.sql.expression.FunctionElement):
    type = sa.types.DateTime()


@sa.ext.compiler.compiles(utcnow, "postgresql")
def pg_utcnow(element, compiler, **kw):
    return "TIMEZONE('utc', CURRENT_TIMESTAMP)"


def temporary_ics_ids():
    """Generator that returns the full list of temporary ICS ids"""
    return (
        f'{current_app.config["TEMPORARY_ICS_ID"]}{letter}{number:0=3d}'
        for letter in string.ascii_uppercase
        for number in range(0, 1000)
    )


def used_temporary_ics_ids():
    """Return a set with the temporary ICS ids used"""
    temporary_items = Item.query.filter(
        Item.ics_id.startswith(current_app.config["TEMPORARY_ICS_ID"])
    ).all()
    return {item.ics_id for item in temporary_items}


def get_temporary_ics_id():
    """Return a temporary ICS id that is available"""
    used_temp_ics_ids = used_temporary_ics_ids()
    for ics_id in temporary_ics_ids():
        if ics_id not in used_temp_ics_ids:
            return ics_id
    else:
        raise ValueError("No temporary ICS id available")


@login_manager.user_loader
def load_user(user_id):
    """User loader callback for flask-login

    :param str user_id: unicode ID of a user
    :returns: corresponding user object or None
    """
    return User.query.get(int(user_id))


@ldap_manager.save_user
def save_user(dn, username, data, memberships):
    """User saver for flask-ldap3-login

    This method is called whenever a LDAPLoginForm()
    successfully validates.
    """
    user = User.query.filter_by(username=username).first()
    if user is None:
        user = User(
            username=username,
            display_name=utils.attribute_to_string(data["cn"]) or username,
            email=utils.attribute_to_string(data["mail"]),
        )
    # Always update the user groups to keep them up-to-date
    user.groups = sorted(
        [utils.attribute_to_string(group["cn"]) for group in memberships]
    )
    db.session.add(user)
    db.session.commit()
    return user


# Tables required for Many-to-Many relationships between users and favorites attributes
favorite_manufacturers_table = db.Table(
    "favorite_manufacturers",
    db.Column(
        "user_id", db.Integer, db.ForeignKey("user_account.id"), primary_key=True
    ),
    db.Column(
        "manufacturer_id",
        db.Integer,
        db.ForeignKey("manufacturer.id"),
        primary_key=True,
    ),
)
favorite_models_table = db.Table(
    "favorite_models",
    db.Column(
        "user_id", db.Integer, db.ForeignKey("user_account.id"), primary_key=True
    ),
    db.Column("model_id", db.Integer, db.ForeignKey("model.id"), primary_key=True),
)
favorite_locations_table = db.Table(
    "favorite_locations",
    db.Column(
        "user_id", db.Integer, db.ForeignKey("user_account.id"), primary_key=True
    ),
    db.Column(
        "location_id", db.Integer, db.ForeignKey("location.id"), primary_key=True
    ),
)
favorite_statuses_table = db.Table(
    "favorite_statuses",
    db.Column(
        "user_id", db.Integer, db.ForeignKey("user_account.id"), primary_key=True
    ),
    db.Column("status_id", db.Integer, db.ForeignKey("status.id"), primary_key=True),
)
favorite_actions_table = db.Table(
    "favorite_actions",
    db.Column(
        "user_id", db.Integer, db.ForeignKey("user_account.id"), primary_key=True
    ),
    db.Column("action_id", db.Integer, db.ForeignKey("action.id"), primary_key=True),
)


class User(db.Model, UserMixin):
    # "user" is a reserved word in postgresql
    # so let's use another name
    __tablename__ = "user_account"

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.Text, nullable=False, unique=True)
    display_name = db.Column(db.Text, nullable=False)
    email = db.Column(db.Text)
    groups = db.Column(postgresql.ARRAY(db.Text), default=[])
    tokens = db.relationship("Token", backref="user")
    tasks = db.relationship("Task", backref="user")
    # The favorites won't be accessed very often so we load them
    # only when necessary (lazy=True)
    favorite_manufacturers = db.relationship(
        "Manufacturer",
        secondary=favorite_manufacturers_table,
        lazy=True,
        backref=db.backref("favorite_users", lazy=True),
    )
    favorite_models = db.relationship(
        "Model",
        secondary=favorite_models_table,
        lazy=True,
        backref=db.backref("favorite_users", lazy=True),
    )
    favorite_locations = db.relationship(
        "Location",
        secondary=favorite_locations_table,
        lazy=True,
        backref=db.backref("favorite_users", lazy=True),
    )
    favorite_statuses = db.relationship(
        "Status",
        secondary=favorite_statuses_table,
        lazy=True,
        backref=db.backref("favorite_users", lazy=True),
    )
    favorite_actions = db.relationship(
        "Action",