decorators.py

from flask import abort, current_app, request
from flask_login import current_user
from functools import wraps
from threading import Thread
from typing import List, Union
from werkzeug.exceptions import NotAcceptable
from app.models import Permission


def permission_required(permission):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if not current_user.can(permission):
                abort(403)
            return f(*args, **kwargs)
        return decorated_function
    return decorator


def admin_required(f):
    return permission_required(Permission.ADMINISTRATE)(f)


def socketio_login_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if current_user.is_authenticated:
            return f(*args, **kwargs)
        else:
            return {'code': 401, 'msg': 'Unauthorized'}
    return decorated_function


def socketio_permission_required(permission):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if not current_user.can(permission):
                return {'code': 403, 'msg': 'Forbidden'}
            return f(*args, **kwargs)
        return decorated_function
    return decorator


def socketio_admin_required(f):
    return socketio_permission_required(Permission.ADMINISTRATE)(f)


def background(f):
    '''
    ' This decorator executes a function in a Thread.
    ' Decorated functions need to be executed within a code block where an
    ' app context exists.
    '
    ' NOTE: An app object is passed as a keyword argument to the decorated
    '       function.
    '''
    @wraps(f)
    def wrapped(*args, **kwargs):
        kwargs['app'] = current_app._get_current_object()
        thread = Thread(target=f, args=args, kwargs=kwargs)
        thread.start()
        return thread
    return wrapped


def consumes(mime_type: str, *_mime_types: str):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            provided = request.mimetype
            consumeables = {mime_type, *_mime_types}
            if provided not in consumeables:
                raise NotAcceptable()
            return f(*args, **kwargs)
        return decorated_function
    return decorator


def produces(mime_type: str, *_mime_types: str):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            accepted = {*request.accept_mimetypes.values()}
            produceables = {mime_type, *_mime_types}
            if len(produceables & accepted) == 0:
                raise NotAcceptable()
            return f(*args, **kwargs)
        return decorated_function
    return decorator


def content_negotiation(
    produces: Union[str, List[str]],
    consumes: Union[str, List[str]]
):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            provided = request.mimetype
            if isinstance(consumes, str):
                consumeables = {consumes}
            else:
                consumeables = {*consumes}
            accepted = {*request.accept_mimetypes.values()}
            if isinstance(produces, str):
                produceables = {produces}
            else:
                produceables = {*produces}
            if len(produceables & accepted) == 0:
                raise NotAcceptable()
            if provided not in consumeables:
                raise NotAcceptable()
            return f(*args, **kwargs)
        return decorated_function
    return decorator