Fork of https://gist.github.com/guyzmo/603ccdd0f504c63cd0df

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_login import LoginManager, UserMixin, current_user, login_required, user_logged_in, user_logged_out
from flask_principal import UserNeed, RoleNeed, Principal, Identity, AnonymousIdentity, identity_changed, \
    identity_loaded

#########
# Flask

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


#########
# Login

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):
    # Try to login using Basic Auth
    # http://flask.pocoo.org/snippets/8/
    auth = request.authorization

    if auth:
        user = User(auth.username)  # XXX consider that this user may not exist
        if auth.password == user.get_password():
            return user

    # return None if no user was authenticated
    return None

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

#########
# Principals


principals = Principal(app)

@principals.identity_loader
def read_identity_from_flask_login():
    if current_user.is_authenticated():
        return Identity(current_user.id)
    return AnonymousIdentity()


@identity_loaded.connect_via(app)
def on_identity_loaded(sender, identity):
    print("on_identity_loaded", identity.id)
    # Add the UserNeed to the identity
    if hasattr(current_user, 'id'):
        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',  # NOTE since rules cascade this is equivalent to simply: {"read": "admin"}
            'create': 'admin',
            'update': 'admin',
            'delete': 'admin'
        }


api = PresstApi(app)

api.decorators = [login_required]

api.add_resource(BookResource)

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