# (c) cavaliba.com - data - group.py # v4.0 - Wrapper around Instance(schema = group) import yaml from app_data.data import Instance # 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 Group(Instance): CLASSNAME = "group" MAX_COMPUTED_USER = 500 def __init__(self, iobj=None, keyname=None, classname=None, expand=False): super().__init__(classname=self.CLASSNAME, iobj=iobj, keyname=keyname, expand=expand) # list of object (unique) self.subgroups = None self.direct_users = None self.auto_users = None self.users = None # no recurse, direct + auto of self self.users_all = None # all users, self + recurse subgroups, direct + auto if expand: _ = self.get_users_all() def __str__(self): return f"{self.keyname}" # ----------------------------------- # Subgroups / Users hierarchy # ----------------------------------- def get_subgroups(self) -> list["Group"]: """Return all transitive subgroups of self via BFS on the subgroups field.""" seen: set[str] = set() self.subgroups = [] queue: list[Group] = [self] queued_ids: set[int] = {self.id} while queue: current = queue.pop(0) field = current.fields.get("subgroups") if not field: continue for keyname in field.value: if not keyname or keyname in seen: continue g = Group.from_keyname(keyname=keyname) if g and g.is_bound: self.subgroups.append(g) seen.add(keyname) if g.id not in queued_ids: queued_ids.add(g.id) queue.append(g) return self.subgroups def get_direct_users(self) -> list["User"]: """Return User objects listed in the users field of this group, no tree walk.""" from app_data.user import User seen: set[str] = set() self.direct_users = [] field = self.fields.get("users") if not field: return self.direct_users for ulogin in field.value: if not ulogin or ulogin in seen: continue u = User.from_keyname(keyname=ulogin) if u and u.is_bound: self.direct_users.append(u) seen.add(ulogin) if len(self.direct_users) >= self.MAX_COMPUTED_USER: break return self.direct_users def get_auto_users(self) -> list["User"]: """Return User objects listed in autogroup_users of this group, no tree walk.""" from app_data.user import User seen: set[str] = set() self.auto_users = [] field = self.fields.get("autogroup_users") if not field: return self.auto_users for ulogin in field.value: if not ulogin or ulogin in seen: continue u = User.from_keyname(keyname=ulogin) if u and u.is_bound: self.auto_users.append(u) seen.add(ulogin) if len(self.auto_users) >= self.MAX_COMPUTED_USER: break return self.auto_users def get_users(self) -> list["User"]: """Return all users (direct + auto) of this group only, no subgroup recursion.""" if self.direct_users is None: _ = self.get_direct_users() if self.auto_users is None: _ = self.get_auto_users() seen: set[str] = set() self.users = [] for u in self.direct_users + self.auto_users: if u.keyname not in seen: seen.add(u.keyname) self.users.append(u) if len(self.users) >= self.MAX_COMPUTED_USER: break return self.users def get_users_all(self) -> list["User"]: """Return all users (direct + auto) of this group and all transitive subgroups, deduplicated.""" if self.subgroups is None: _ = self.get_subgroups() seen: set[str] = set() self.users_all = [] for u in self.get_users(): if u.keyname not in seen: seen.add(u.keyname) self.users_all.append(u) if len(self.users_all) >= self.MAX_COMPUTED_USER: return self.users_all for g in self.subgroups: for u in g.get_users(): if u.keyname not in seen: seen.add(u.keyname) self.users_all.append(u) if len(self.users_all) >= self.MAX_COMPUTED_USER: return self.users_all return self.users_all # ----------------------------------- # Autogroup # ----------------------------------- @classmethod def autogroup_update(cls, groups=None): """ Recompute autogroup_users for each group and persist to DB. groups: list of Group objects; if None or empty, processes all enabled group instances. Returns count of groups that yielded at least one computed user. """ count = 0 if not groups: groups = cls.iterate_classname(enabled="yes") for group in groups: users = group.get_users_computed() if users: count += 1 group.fields["autogroup_users"].value = [u.keyname for u in users] group.update() return count def get_users_computed(self): """ Parse self.autogroup grammar (schema:keyname:fieldname) and return list of User objects. G1 (keyname=*): scan all instances of schema via DataEAV. G2 (fieldname=*): scan all user fields of one specific instance. Returns [] if autogroup is empty or grammar is invalid. """ # from app_data.user import User autogroup = self.get_attribute_first("autogroup") if not autogroup: return [] try: schema_name, keyname, fieldname = autogroup.split(":") except ValueError: return [] if keyname == "*": return self.get_users_computed_g1(schema_name, fieldname) elif fieldname == "*": return self.get_users_computed_g2(schema_name, keyname) return [] def get_users_computed_g1(self, schema_name, fieldname): """ G1 grammar: schema_name:*:fieldname Query DataEAV for all (classname=schema_name, fieldname=fieldname) rows. Each row value is a user keyname. Returns deduplicated list of User objects. """ # DataEAV scan, no instance loading from app_data.models import DataEAV from app_data.user import User reply = [] seen = set() for eav in DataEAV.objects.filter(classname=schema_name, fieldname=fieldname): ulogin = eav.value if not ulogin or ulogin in seen: continue u = User.from_keyname(keyname=ulogin) if u and u.is_bound: reply.append(u) seen.add(ulogin) return reply def get_users_computed_g2(self, schema_name, keyname): """ G2 grammar: schema_name:keyname:* Load instance schema_name:keyname and collect values from every field whose dataformat_ext points to user. Returns deduplicated list of User objects. """ # one instance, all fields pointing to user from app_data.data import Instance from app_data.user import User reply = [] seen = set() instance = Instance.from_keyname(classname=schema_name, keyname=keyname) if not instance or not instance.is_bound: return reply for field in instance.fields.values(): if field.dataformat != "schema" or field.dataformat_ext.split()[0] != "user": continue for ulogin in field.value: if ulogin in seen: continue u = User.from_keyname(keyname=ulogin) if u and u.is_bound: reply.append(u) seen.add(ulogin) return reply # --- inherited from Instance() ---