Source code for parameter_info.cached_info

# Copyright (c) Fraunhofer MEVIS, Germany. All rights reserved.
# **InsertLicense** code author="Jan-Martin Kuhnigk"

from parameter_info.indirect_info import IndirectParameterInfo, InformationSource, Query, \
    raise_InformationSourceKeyError


class CachedParameterInfo(IndirectParameterInfo):
    """
      Extension of IndirectParameterInfo providing a cache for data retrieved from a user-defined information source.
    """
    def __init__(self, default_information_source_query):
        super(CachedParameterInfo, self).__init__(default_information_source_query)
        assert issubclass( default_information_source_query.query_class, CachedQuery )

    def clear_cached_values(self ):
        """
        Clears all cached values, forcing to update from source on next request. Works recursively, as long
        as the child CachedParameterInfo is a direct child.

        :return: None
        """
        for key in self:
            if isinstance( self._get_internal_value( key ), CachedParameterInfo ):
                self._get_internal_value( key ).clear_cached_values()
            else:
                self._get_internal_value( key ).clear()

    # THIS IS THE VERSION THAT SHOULD WORK FULLY RECURSIVELY
    # TODO: Add tests, check if we really need to differentiate between IndirectParamInfo and dict when descending
    # def clear_cached_values(self ):
    #     """
    #     Clears all cached values, forcing to update from source on next request. Works recursively, as long
    #     as the child CachedParameterInfo is a direct child.
    #     :return: None
    #     """
    #     self.clear_cached_values_recursively( self )
    #
    # @classmethod
    # def clear_cached_values_recursively( cls, d ):
    #     for key in d:
    #         is_child_node = False
    #         if isinstance( d, IndirectParameterInfo ):
    #             if isinstance( d._get_internal_value( key ), CachedQuery ):
    #                 d._get_internal_value( key ).clear()
    #             if not isinstance( d._get_internal_value( key ), Query ) and isinstance( d[key], dict ):
    #                 is_child_node = True
    #         elif isinstance( d[ key ], dict ):
    #             is_child_node = True
    #         if is_child_node:
    #             cls.clear_cached_values_recursively( d[ key ] )

    def update_cache_from_source( self ):
        """
        Refreshes (overwrites) all (possibly cached) values with those from the information source.
        Works recursively, as long as the child CachedParameterInfo is a direct child.

        :return: None
        """
        for key in self:
            if isinstance( self._get_internal_value( key ), CachedParameterInfo ):
                self._get_internal_value( key ).update_cache_from_source()
            else:
                self._get_internal_value( key ).update()


class CachedInformationSource( InformationSource ):
    """
    Convenience class, just exchanging the query_class with one supporting caching.
    """

    def __init__(self, source_key_failure_handler=raise_InformationSourceKeyError,
                 **source_query_data):
        """
        Creates a InformationSource with a CachedQuery as query_class.
        See super class documentation for parameter info.
        """
        super( CachedInformationSource, self ).__init__( source_key_failure_handler=source_key_failure_handler,
                                                         **source_query_data )
        self.query_class = CachedQuery


class CachedQuery(Query):
    def __init__(self, getter_cb, *getter_args, **getter_kwargs):
        """
        See super class for parameter documentation
        """
        super(CachedQuery, self).__init__( getter_cb, *getter_args, **getter_kwargs)
        self.__has_cached_value = False
        self.__cached_value = None

    def __call__(self):
        """
        Executes the query, returning result value.

        :return: The cached value if existing, otherwise updates the cache first.
        """
        if not self.__has_cached_value:
            self.update()
        return self.__cached_value

    def clear(self):
        """
        Clears the cache

        :return: None
        """
        self.__has_cached_value = False
        self.__cached_value = None

    def update(self):
        """
        Refreshes the cache from the source.

        :return: None
        """
        self.__cached_value = super(CachedQuery, self).__call__()
        self.__has_cached_value = True

    def _has_cached_value(self):
        """
        Currently only required for testing, thus protected.

        :return: True if a cached value exists
        """
        return self.__has_cached_value