116 lines
3.9 KiB
Python
116 lines
3.9 KiB
Python
from ..database import db
|
|
from sqlalchemy.orm.collections import attribute_mapped_collection
|
|
from flask import current_app
|
|
from werkzeug.local import LocalProxy
|
|
|
|
logger = LocalProxy(lambda: current_app.logger)
|
|
|
|
association_table = db.Table('user_x_role',
|
|
db.Column('user_id', db.Integer, db.ForeignKey('user.id')),
|
|
db.Column('role_id', db.Integer, db.ForeignKey('role.id'))
|
|
)
|
|
|
|
|
|
class User(db.Model):
|
|
""" Database Object for User
|
|
|
|
Table for all saved User
|
|
|
|
Attributes:
|
|
id: Id in Database as Primary Key.
|
|
uid: User ID used by authentication provider
|
|
display_name: Name to show
|
|
firstname: Firstname of the User
|
|
lastname: Lastname of the User
|
|
mail: mail address of the User
|
|
"""
|
|
__tablename__ = 'user'
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
uid = db.Column(db.String(30))
|
|
display_name = db.Column(db.String(30))
|
|
firstname = db.Column(db.String(30))
|
|
lastname = db.Column(db.String(30))
|
|
mail = db.Column(db.String(30))
|
|
roles = db.relationship("Role", secondary=association_table)
|
|
sessions = db.relationship("AccessToken", back_populates="user")
|
|
attributes = db.relationship("UserAttribute", collection_class=attribute_mapped_collection('name'),
|
|
cascade="all, delete")
|
|
|
|
def set_attribute(self, name, value):
|
|
if name in self.attributes:
|
|
self.attributes[name].value = value
|
|
else:
|
|
self.attributes[name] = UserAttribute(name=name, value=value)
|
|
|
|
def add_role(self, name):
|
|
r = Role.query.filter_by(name=name).first()
|
|
if not r:
|
|
r = Role(name=name)
|
|
self.roles.append(r)
|
|
|
|
def update_data(self, data):
|
|
logger.debug("update data of user")
|
|
if 'uid' in data:
|
|
self.uid = data['uid']
|
|
if 'firstname' in data:
|
|
self.firstname = data['firstname']
|
|
if 'lastname' in data:
|
|
self.lastname = data['lastname']
|
|
if 'mail' in data:
|
|
self.mail = data['mail']
|
|
if 'display_name' in data:
|
|
self.display_name = data['display_name']
|
|
|
|
def get_permissions(self):
|
|
return ["user"] + [permission.name for role in self.roles for permission in role.permissions]
|
|
|
|
def has_permissions(self, permissions):
|
|
for role in self.roles:
|
|
for permission in role.permissions:
|
|
if permission.name in permissions:
|
|
return True
|
|
return False
|
|
|
|
def serialize(self):
|
|
return {
|
|
"userid": self.uid,
|
|
"display_name": self.display_name,
|
|
"firstname": self.firstname,
|
|
"lastname": self.lastname,
|
|
"mail": self.mail,
|
|
"roles": [r.name for r in self.roles]
|
|
}
|
|
|
|
|
|
class UserAttribute(db.Model):
|
|
__tablename__ = 'user_attribute'
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
user = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
|
|
name = db.Column(db.String(30))
|
|
value = db.Column(db.String(192))
|
|
|
|
|
|
role_permission_association_table = db.Table('role_x_permission',
|
|
db.Column('role_id', db.Integer, db.ForeignKey('role.id')),
|
|
db.Column('permission_id', db.Integer, db.ForeignKey('permission.id'))
|
|
)
|
|
|
|
|
|
class Role(db.Model):
|
|
__tablename__ = 'role'
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(30), unique=True)
|
|
permissions = db.relationship("Permission", secondary=role_permission_association_table, cascade="all, delete")
|
|
|
|
def serialize(self):
|
|
return self.name
|
|
|
|
|
|
class Permission(db.Model):
|
|
__tablename__ = 'permission'
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(30), unique=True)
|
|
|
|
def serialize(self):
|
|
return self.name
|