# (c) cavaliba.com - data - user.py # v4.0 - Wrapper around Instance(schema = user) import yaml from django.utils import timezone from django.utils.timezone import localtime from app_data.data import Instance from app_data.group import Group from app_data.models import DataEAV from app_data.role import Role # YAML class MyYamlDumper(yaml.SafeDumper): def write_line_break(self, data=None): super().write_line_break(data) if len(self.indents) < 2: super().write_line_break() class User(Instance): CLASSNAME = "user" def __init__(self, iobj=None, keyname=None, classname=None, expand=False): super().__init__(classname=self.CLASSNAME, iobj=iobj, keyname=keyname, expand=expand) # list of objects (unique) self.direct_groups = None self.parent_groups = None self.auto_groups = None self.groups = None self.direct_roles = None self.roles = None self.permissions = None # if expand: # self.compute_aaa() def __str__(self): return f"{self.keyname}" # ----------------------------------- # User specific # ----------------------------------- @classmethod def get_by_email(cls, email=None): if not email: return None eav = DataEAV.objects.filter( classname=cls.CLASSNAME, fieldname="email", value=email ).first() if not eav: return None return cls.from_id(id=eav.iid) @property def login(self): return self.keyname @property def last_login(self): return self.get_attribute_first("last_login") @property def email(self): return self.get_attribute_first("email") def get_email(self): secondary = self.get_attribute_first("secondary_email") if secondary: return secondary return self.get_attribute_first("email") def get_mobile(self): secondary = self.get_attribute_first("secondary_mobile") if secondary: return secondary return self.get_attribute_first("mobile") def send_test_email(self, aaa=None): from app_sirene.notify import send_test_email send_test_email(user=self, aaa=aaa) def send_test_sms(self, aaa=None): from app_sirene.notify import send_test_sms return send_test_sms(user=self, aaa=aaa) @property def want_notifications(self) -> bool: return self.is_field_true("want_notifications") @property def want_24(self) -> bool: return self.is_field_true("want_24") # def to_json(self): # data = self.to_dict() # return json.dumps(data, indent=4, ensure_ascii=False).encode('utf8') # def to_yaml(self): # data = self.to_dict() # return yaml.dump(data, allow_unicode=True, Dumper=MyYamlDumper, sort_keys=False) # -------------------------------------- # USER/GROUP/ROLE/PERMIISION hierarchy # -------------------------------------- # user > direct_groups # direct_groups > parent_groups # user > auto_groups # *_groups > groups # groups > roles # user > roles # roles > permissions # -------------------------------------- def compute_aaa(self): """ populate if None: - self.direct_groups - self.parent_groups - self.auto_groups - self.direct_roles - self.roles - self.permissions """ _ = self.get_direct_groups() _ = self.get_parent_groups() _ = self.get_auto_groups() _ = self.get_groups() _ = self.get_roles() _ = self.get_permissions() def get_permissions(self): from app_data.permissions import permission_get_all if self.roles is None: _ = self.get_roles() seen = set() self.permissions = [] if self.keyname == "admin": self.permissions = list(permission_get_all()) return self.permissions for role in self.roles: if role.keyname == "role_admin": self.permissions = list(permission_get_all()) return self.permissions for p in role.get_permissions(): if p.keyname not in seen: self.permissions.append(p) seen.add(p.keyname) role_default = Role.from_keyname(keyname="role_default") if role_default and role_default.is_bound: for p in role_default.get_permissions(): if p.keyname not in seen: self.permissions.append(p) seen.add(p.keyname) return self.permissions def get_roles(self): seen = set() self.roles = [] direct_eav = DataEAV.objects.filter(classname="role", fieldname="users", value=self.keyname) if self.groups is None: _ = self.get_groups() group_keynames = [g.keyname for g in self.groups] group_eav = DataEAV.objects.filter( classname="role", fieldname="groups", value__in=group_keynames ) for eav in list(direct_eav) + list(group_eav): if eav.iid in seen: continue r = Role.from_id(id=eav.iid) if r and r.is_bound: self.roles.append(r) seen.add(eav.iid) return self.roles def get_groups(self): if self.direct_groups is None: _ = self.get_direct_groups() if self.parent_groups is None: _ = self.get_parent_groups() if self.auto_groups is None: _ = self.get_auto_groups() seen = {} self.groups = [] for g in self.direct_groups + self.parent_groups + self.auto_groups: if g.keyname not in seen: seen[g.keyname] = g self.groups.append(g) return self.groups def get_direct_groups(self): eav_qs = DataEAV.objects.filter(classname="group", fieldname="users", value=self.keyname) self.direct_groups = [] for eav in eav_qs: g = Group.from_id(id=eav.iid) if g and g.is_bound: self.direct_groups.append(g) return self.direct_groups def get_auto_groups(self): """ return groups where self user in autogroup """ eav_qs = DataEAV.objects.filter( classname="group", fieldname="autogroup_users", value=self.keyname ) self.auto_groups = [] for eav in eav_qs: g = Group.from_id(id=eav.iid) if g and g.is_bound: self.auto_groups.append(g) return self.auto_groups def get_parent_groups(self): if self.direct_groups is None: _ = self.get_direct_groups() direct_keynames = {g.keyname for g in self.direct_groups} self.parent_groups = [] seen = set(direct_keynames) redo = 0 for g0 in self.direct_groups: for eav in DataEAV.objects.filter( classname="group", fieldname="subgroups", value=g0.keyname ): g1 = Group.from_id(id=eav.iid) if g1 and g1.is_bound and g1.keyname not in seen: self.parent_groups.append(g1) seen.add(g1.keyname) redo = 1 while redo > 0: redo = 0 for g1 in list(self.parent_groups): for eav in DataEAV.objects.filter( classname="group", fieldname="subgroups", value=g1.keyname ): g2 = Group.from_id(id=eav.iid) if g2 and g2.is_bound and g2.keyname not in seen: self.parent_groups.append(g2) seen.add(g2.keyname) redo = 1 return self.parent_groups def update_last_login(self): now_str = localtime(timezone.now()).strftime("%Y-%m-%d %H:%M:%S") self.set_field_value_single(fieldname="last_login", value=now_str) self.update()