def _fill_role_cache_batch(self, principals, overwrite=False):
        """Fill role cache for `principals` (Users and/or Groups), in order to
        avoid too many queries when checking role access with 'has_role'."""
        if not self.app_state.use_cache:
            return

        query = db.session.query(RoleAssignment)
        users = {u for u in principals if isinstance(u, User)}
        groups = {g for g in principals if isinstance(g, Group)}
        groups |= {g for u in users for g in u.groups}

        if not overwrite:
            users = {u for u in users if not self._has_role_cache(u)}
            groups = {g for g in groups if not self._has_role_cache(g)}

        if not (users or groups):
            return

        # ensure principals processed here will have role cache. Thus users or
        # groups without any role will have an empty role cache, to avoid
        # unneeded individual DB query when calling self._fill_role_cache(p).
        for p in chain(users, groups):
            self._set_role_cache(p, {})

        filter_cond = []
        if users:
            filter_cond.append(RoleAssignment.user_id.in_(u.id for u in users))
        if groups:
            filter_cond.append(RoleAssignment.group_id.in_(g.id for g in groups))

        query = query.filter(sql.or_(*filter_cond))
        ra_users = {}
        ra_groups = {}
        for ra in query.all():
            if ra.user:
                all_roles = ra_users.setdefault(ra.user, {})
            else:
                all_roles = ra_groups.setdefault(ra.group, {})

            object_key = (
                f"{ra.object.entity_type}:{ra.object_id:d}"
                if ra.object is not None
                else None
            )
            all_roles.setdefault(object_key, set()).add(ra.role)

        for group, all_roles in ra_groups.items():
            self._set_role_cache(group, all_roles)

        for user, all_roles in ra_users.items():
            for gr in user.groups:
                group_roles = self._fill_role_cache(gr)
                for object_key, roles in group_roles.items():
                    obj_roles = all_roles.setdefault(object_key, set())
                    obj_roles |= roles

            self._set_role_cache(user, all_roles)
def _fill_role_cache_batch ( self , principals , overwrite = False ) : if not self . app_state . use_cache : return query = db . session . query ( RoleAssignment ) users = { u for u in principals if isinstance ( u , User ) } groups = { g for g in principals if isinstance ( g , Group ) } groups |= { g for u in users for g in u . groups } if not overwrite : users = { u for u in users if not self . _has_role_cache ( u ) } groups = { g for g in groups if not self . _has_role_cache ( g ) } if not ( users or groups ) : return for p in chain ( users , groups ) : self . _set_role_cache ( p , { } ) filter_cond = [ ] if users : filter_cond . append ( RoleAssignment . user_id . in_ ( u . id for u in users ) ) if groups : filter_cond . append ( RoleAssignment . group_id . in_ ( g . id for g in groups ) ) query = query . filter ( sql . or_ ( * filter_cond ) ) ra_users = { } ra_groups = { } for ra in query . all ( ) : if ra . user : all_roles = ra_users . setdefault ( ra . user , { } ) else : all_roles = ra_groups . setdefault ( ra . group , { } ) object_key = ( f"{ra.object.entity_type}:{ra.object_id:d}" if ra . object is not None else None ) all_roles . setdefault ( object_key , set ( ) ) . add ( ra . role ) for group , all_roles in ra_groups . items ( ) : self . _set_role_cache ( group , all_roles ) for user , all_roles in ra_users . items ( ) : for gr in user . groups : group_roles = self . _fill_role_cache ( gr ) for object_key , roles in group_roles . items ( ) : obj_roles = all_roles . setdefault ( object_key , set ( ) ) obj_roles |= roles self . _set_role_cache ( user , all_roles )
Fill role cache for `principals` (Users and/or Groups), in order to
        avoid too many queries when checking role access with 'has_role'.