Skip to content
Snippets Groups Projects
decorators.py 1.74 KiB
Newer Older
  • Learn to ignore specific revisions
  • Patrick Jentsch's avatar
    Patrick Jentsch committed
    from flask import abort, current_app
    
    from flask_login import current_user
    
    from functools import wraps
    
    from threading import Thread
    
    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
    
        return permission_required(Permission.ADMINISTRATE)(f)
    
    
    def socketio_login_required(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)
    
    Patrick Jentsch's avatar
    Patrick Jentsch committed
        '''
        ' 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()