Method search

msal/token_cache.py:236–289

Returns a generator of matching entries. It is O(1) for AT hits, and O(n) for other types. Note that it holds a lock during the entire search.

(self, credential_type, target=None, query=None, *, now=None)

            if target_set else True)

    def search(self, credential_type, target=None, query=None, *, now=None):  # O(n) generator
        """Returns a generator of matching entries.

        It is O(1) for AT hits, and O(n) for other types.
        Note that it holds a lock during the entire search.
        """
        target = sorted(target or [])  # Match the order sorted by add()
        assert isinstance(target, list), "Invalid parameter type"

        preferred_result = None
        if (credential_type == self.CredentialType.ACCESS_TOKEN
            and isinstance(query, dict)
            and "home_account_id" in query and "environment" in query
            and "client_id" in query and "realm" in query and target
        ):  # Special case for O(1) AT lookup
            preferred_result = self._get_access_token(
                query["home_account_id"], query["environment"],
                query["client_id"], query["realm"], target,
                ext_cache_key=query.get("ext_cache_key"))
            if preferred_result and self._is_matching(
                preferred_result, query,
                # Needs no target_set here because it is satisfied by dict key
            ):
                yield preferred_result

        target_set = set(target)
        with self._lock:
            # O(n) search. The key is NOT used in search.
            now = int(time.time() if now is None else now)
            expired_access_tokens = [
                # Especially when/if we key ATs by ephemeral fields such as key_id,
                # stale ATs keyed by an old key_id would stay forever.
                # Here we collect them for their removal.
            ]
            for entry in self._cache.get(credential_type, {}).values():
                if (  # Automatically delete expired access tokens
                    credential_type == self.CredentialType.ACCESS_TOKEN
                    and int(entry["expires_on"]) < now
                ):
                    expired_access_tokens.append(entry)  # Can't delete them within current for-loop
                    continue
                if (entry != preferred_result  # Avoid yielding the same entry twice
                    and self._is_matching(entry, query, target_set=target_set)
                ):
                    # Cache isolation for extended cache keys (e.g., FMI path).
                    # Entries with ext_cache_key must not match queries without one.
                    if (credential_type == self.CredentialType.ACCESS_TOKEN
                        and "ext_cache_key" in entry
                        and "ext_cache_key" not in (query or {})
                    ):
                        continue
                    yield entry
            for at in expired_access_tokens:
                self.remove_at(at)

    def find(self, credential_type, target=None, query=None, *, now=None):
        """Equivalent to list(search(...))."""