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 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 content_negotiation( produces: Union[str, List[str], None] = None, consumes: Union[str, List[str], None] = None ): def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): provided = request.mimetype if consumes is None: consumeables = None elif isinstance(consumes, str): consumeables = {consumes} elif isinstance(consumes, list) and all(isinstance(x, str) for x in consumes): consumeables = {*consumes} else: raise TypeError() accepted = {*request.accept_mimetypes.values()} if produces is None: produceables = None elif isinstance(produces, str): produceables = {produces} elif isinstance(produces, list) and all(isinstance(x, str) for x in produces): produceables = {*produces} else: raise TypeError() if produceables is not None and len(produceables & accepted) == 0: raise NotAcceptable() if consumeables is not None and provided not in consumeables: raise NotAcceptable() return f(*args, **kwargs) return decorated_function return decorator