mirror of
https://gitlab.ub.uni-bielefeld.de/sfb1288inf/nopaque.git
synced 2024-11-15 17:25:44 +00:00
100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
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 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
|