def _get_most_recent_version()

in src/dynamodb_encryption_sdk/material_providers/most_recent.py [0:0]

• 31 lines of code
• 7 McCabe index (conditional complexity)

    def _get_most_recent_version(self, ttl_action):
        # type: (bool) -> CryptographicMaterialsProvider
        """Get the most recent version of the provider.

        If allowing local and we cannot obtain the lock, just return the most recent local
        version. Otherwise, wait for the lock and ask the provider store for the most recent
        version of the provider.

        :param TtlActions ttl_action: The ttl action to take for this version
        :returns: version and corresponding cryptographic materials provider
        :rtype: CryptographicMaterialsProvider
        """
        blocking_wait = bool(ttl_action is TtlActions.EXPIRED)
        acquired = self._lock.acquire(blocking_wait)  # pylint: disable=consider-using-with

        if not acquired:
            # We failed to acquire the lock.
            # If blocking, we will never reach this point.
            # If not blocking, we want whatever the latest local version is.
            _LOGGER.debug("Failed to acquire lock. Returning the last cached version.")
            version = self._version
            _, provider = self._cache.get(version)
            return provider

        try:
            # If the entry was expired then we blocked waiting for the lock, so it's possible some other thread already
            # queried the provider store and re-populated the cache. If so, we don't want to re-query the provider
            # store, so check if the entry is back in the cache first
            if ttl_action is TtlActions.EXPIRED:
                try:
                    _, provider = self._cache.get(self._version)
                    return provider
                except KeyError:
                    pass

            max_version = self._get_max_version()
            try:
                _, provider = self._cache.get(max_version)
            except KeyError:
                provider = self._get_provider(max_version)
            received_version = self._provider_store.version_from_material_description(
                provider._material_description  # pylint: disable=protected-access
            )

            _LOGGER.debug("Caching materials provider version %d", received_version)
            self._version = received_version  # pylint: disable=attribute-defined-outside-init
            self._last_updated = time.time()  # pylint: disable=attribute-defined-outside-init
            self._cache.put(received_version, (self._last_updated, provider))
        finally:
            self._lock.release()

        _LOGGER.debug("New latest version is %d", self._version)

        return provider