Simple example of Flask/Presst with Login and Principals (not working!)

simple-example-login-principals.py

#!/usr/bin/env python

from base64 import b64decode

from flask import Flask, current_app
from flask_sqlalchemy import SQLAlchemy
from flask_presst import PresstApi, ModelResource
from flask_presst.principal import PrincipalResource
from flask.ext.principal import UserNeed, RoleNeed

#########
# Flask

app = Flask(__name__)
app.config.update(dict(
    SQLALCHEMY_DATABASE_URI = 'sqlite://',
    SECRET_KEY = 'secret_xxx'
))


#########
# Login

from flask_login import LoginManager, UserMixin, current_user, login_required

login_manager = LoginManager(app)

class User(UserMixin):
    def __init__(self, id):
        self.id = id
        self.roles = [id]

    def get_password(self):
        return self.id

    def __repr__(self):
        return "{}".format(self.id)

@login_manager.request_loader
def load_user_from_request(request):
    print("load_user_from_request", request)

    # next, try to login using Basic Auth
    api_key = request.headers.get('Authorization')
    if api_key:
        api_key = api_key.replace('Basic ', '', 1)
        try:
            auth = b64decode(api_key).decode()
            if ":" in auth:
                username, password = auth.split(":")
                user = User(username)
                if password == user.get_password():
                    #
                    identity = Identity(user.id)
                    if hasattr(current_user, 'id'):
                        identity.provides.add(UserNeed(current_user.id))
                    if hasattr(current_user, 'roles'):
                        for role in current_user.roles:
                            identity.provides.add(RoleNeed(role))
                    #
                    identity_changed.send(current_app._get_current_object(),
                                          identity=identity)
                    return user
        except Exception as err:
            print("load_user_from_request", err)

    # finally, return None if both methods did not login the user
    return None

@login_manager.user_loader
def load_user(userid):
    print("load_user")
    return User(userid)

#########
# Principals

from flask.ext.principal import Principal, Identity, AnonymousIdentity, identity_changed, identity_loaded

principals = Principal(app)

# @identity_loaded.connect_via(app)
# def on_identity_loaded(sender, identity):
#     print("on_identity_loaded")
#     # Set the identity user object
#     identity.user = current_user
#
#     # Add the UserNeed to the identity
#     if hasattr(current_user, 'id'):
#         print("2")
#         identity.provides.add(UserNeed(current_user.id))
#
#     # Assuming the User model has a list of roles, update the
#     # identity with the roles that the user provides
#     if hasattr(current_user, 'roles'):
#         for role in current_user.roles:
#             identity.provides.add(RoleNeed(role))

#########
# Model

db = SQLAlchemy(app)

class Book(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(), nullable=False)
    year_published = db.Column(db.Integer)

db.create_all()

#########
# Resource

class BookResource(PrincipalResource):
    class Meta:
        model = Book
        permissions = {
            'read': 'admin',
            'create': 'admin',
            'update': 'admin',
            'delete': 'admin'
        }

api = PresstApi(app)

api.decorators = [login_required]

api.add_resource(BookResource)

if __name__ == '__main__':
    app.run()