1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178
| import logging as logger import uuid
from flask import Flask, g from flask_admin import Admin from flask_admin.contrib.sqla import ModelView from flask_curd_db import FlaskCurdDb from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__) app.config["SECRET_KEY"] = str(uuid.uuid4()) app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///rbac.db" app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True app.config['FLASK_ADMIN_SWATCH'] = 'cerulean'
db = SQLAlchemy(app) FlaskCurdDb(app, db, logger)
admin = Admin(app, name='flask-rbac', template_mode='bootstrap3')
from flask_curd_db.db import CRUDMixin
tb_user_role = db.Table( 'tb_user_role', db.Column("role_id", db.Integer, db.ForeignKey("tb_role.id"), nullable=False), db.Column("user_id", db.Integer, db.ForeignKey("tb_user.id"), nullable=False) ) tb_role_menu = db.Table( 'tb_role_menu', db.Column("role_id", db.Integer, db.ForeignKey("tb_role.id"), nullable=False), db.Column("menu_id", db.Integer, db.ForeignKey("tb_menu.id"), nullable=False) ) tb_menu_pemission = db.Table( 'tb_menu_pemission', db.Column("menu_id", db.Integer, db.ForeignKey("tb_menu.id"), nullable=False), db.Column("pemission_id", db.Integer, db.ForeignKey("tb_pemission.id"), nullable=False) ) tb_pemission_operation = db.Table( 'tb_pemission_operation', db.Column("pemission_id", db.Integer, db.ForeignKey("tb_pemission.id"), nullable=False), db.Column("operation_id", db.Integer, db.ForeignKey("tb_operation.id"), nullable=False), )
class Operation(db.Model, CRUDMixin): __tablename__ = "tb_operation" name = db.Column(db.String(50))
def __str__(self): return self.name
class Pemission(db.Model, CRUDMixin): __tablename__ = "tb_pemission" name = db.Column(db.String(50)) tb_pemission_operation = db.relationship(Operation, secondary=tb_pemission_operation, backref=db.backref('pemissions', lazy='dynamic'))
def __str__(self): return self.name
class Menu(db.Model, CRUDMixin): __tablename__ = 'tb_menu' name = db.Column(db.String(50)) tb_menu_pemission = db.relationship(Pemission, secondary=tb_menu_pemission, backref=db.backref('menus', lazy='dynamic'))
def __str__(self): return self.name
class Role(db.Model, CRUDMixin): __tablename__ = "tb_role" name = db.Column(db.String(50)) tb_role_menu = db.relationship(Menu, secondary=tb_role_menu, backref=db.backref('roles', lazy='dynamic'))
def __str__(self): return self.name
class User(db.Model, CRUDMixin): __tablename__ = "tb_user" name = db.Column(db.String(50)) tb_user_role = db.relationship(Role, secondary=tb_user_role, backref=db.backref('users', lazy='dynamic'))
def __str__(self): return self.name
admin.add_view(ModelView(User, db.session)) admin.add_view(ModelView(Role, db.session)) admin.add_view(ModelView(Menu, db.session)) admin.add_view(ModelView(Pemission, db.session)) admin.add_view(ModelView(Operation, db.session))
from functools import wraps
def UserLoginError(BaseException): def __init__(self, message, status): super().__init__(message, status) self.message = message self.status = status
@app.before_request def before_request(): g.user = User.get_ins_by_id(1) print(g.user.name) g.user_roles = g.user.tb_user_role print(g.user_roles) g.user_role_menus = [] for role in g.user_roles: g.user_role_menus.extend(role.tb_role_menu) g.user_role_menus = list(set(g.user_role_menus)) print(g.user_role_menus) g.user_role_menu_pemissions = {} for menu in g.user_role_menus: for pemission in menu.tb_menu_pemission: g.user_role_menu_pemissions[menu.name] = {pemission.name: [operation.name for operation in pemission.tb_pemission_operation]} print(g.user_role_menu_pemissions)
@app.after_request def after_request(reponse): return reponse
def login_require(fn): @wraps(fn) def _dec(*args, **kwargs): if g.user.name == "小黄": return fn(*args, **kwargs) return "滚回去登录!!!"
return _dec
def user_operation_menu(menu_name, pemission_name, operation_name): def user_operation(fn): @wraps(fn) def _dec(*args, **kwargs): if menu_name in g.user_role_menu_pemissions: if pemission_name in g.user_role_menu_pemissions[menu_name]: if operation_name in g.user_role_menu_pemissions[menu_name][pemission_name]: return fn(*args, **kwargs) return "没权限,玩**呢!!!"
return _dec
return user_operation
@app.route('/') @login_require @user_operation_menu("用户管理", "用户管理的权限", "查看") def hello_world(): return 'Hello World!'
if __name__ == '__main__': db.create_all() app.run(debug=True)
|