Skip to content

The client for remote calendars with builtin cache support

When you have a service that periodically runs, and you don't want to fetch a new version of your calendar each time, this is the go-to place.

CacheClient

A iCalendar client which takes care of caching the result for you. This avoids you needing to handle the caching.

Parameters:

Name Type Description Default
cache_location Union[Path, str]

A path to the cache. Can be relative or absolute references. When you pass in a value with a file extension, it is considered to be a directory, otherwise it's considered as a file reference.

required
cache_ttl Union[Duration]

The time-to-live for the cache. The cache will be deleted/refreshed once it is older than the TTL.

Duration(hours=1)
verbose bool

Print verbose messages regarding cache usage.

True
url str

The URL to the iCalendar file.

required
Source code in ical_library/cache_client.py
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class CacheClient:
    """
    A iCalendar client which takes care of caching the result for you.
    This avoids you needing to handle the caching.

    :param cache_location: A path to the cache. Can be relative or absolute references. When you pass in a value with
    a file extension, it is considered to be a directory, otherwise it's considered as a file reference.
    :param cache_ttl: The time-to-live for the cache. The cache will be deleted/refreshed once it is older than the TTL.
    :param verbose: Print verbose messages regarding cache usage.
    :param url: The URL to the iCalendar file.
    """

    def __init__(
        self,
        url: str,
        cache_location: Union[Path, str],
        cache_ttl: Union[Duration] = Duration(hours=1),
        verbose: bool = True,
    ):
        self.url: str = url
        self.cache_location: Path = Path(cache_location)
        self.cache_ttl: Duration = cache_ttl
        self.verbose = verbose

    @property
    def cache_file_path(self) -> Path:
        """Return the filepath to the cache for the given URL."""
        if self.cache_location.suffix == "":
            return self.cache_location / hashlib.md5(self.url.encode()).hexdigest()
        return self.cache_location

    def get_icalendar(self, **kwargs: Any) -> VCalendar:
        """
        Get a parsed VCalendar instance. If there is an active cache, return that, otherwise fetch and cache the result.
        :param kwargs: Any keyword arguments to pass onto the `urllib.request.urlopen` call.
        :return: a VCalendar instance with all it's iCalendar components like VEvents, VToDos, VTimeZones etc.
        """
        if not self._is_cache_expired():
            if self.verbose:
                print("Using cache to remove this folder.")
            return client.parse_icalendar_file(self.cache_file_path)

        self._purge_icalendar_cache()
        response = request.urlopen(self.url, **kwargs)
        if not (200 <= response.getcode() < 400):
            raise ValueError(f"Unable to execute request at {self.url=}. Response code was: {response.getcode()}.")
        text = response.read().decode("utf-8")
        self._write_response_to_cache(text)

        lines = text.split("\n")
        return client.parse_lines_into_calendar(lines)

    def _write_response_to_cache(self, text: str) -> None:
        """
        Write the response of the fetched URL to cache.
        :param text: The fetched result.
        """
        if self.verbose:
            print(f"Successfully loaded new iCalendar data and stored it at {self.cache_file_path}.")
        self.cache_file_path.parent.mkdir(parents=True, exist_ok=True)
        with open(self.cache_file_path, "w") as file:
            file.write(text)

    def _purge_icalendar_cache(self) -> None:
        """Purge the cache we have for this Calendar."""
        if self.verbose:
            print(f"Cache was expired. Removed {self.cache_file_path}.")
        return self.cache_file_path.unlink()

    def _is_cache_expired(self) -> bool:
        """Return whether the cache is passed its expiration date."""
        cutoff = DateTime.utcnow() - self.cache_ttl
        mtime = DateTime.utcfromtimestamp(os.path.getmtime(self.cache_file_path))
        return mtime < cutoff

cache_file_path: Path property

Return the filepath to the cache for the given URL.

_is_cache_expired()

Return whether the cache is passed its expiration date.

Source code in ical_library/cache_client.py
82
83
84
85
86
def _is_cache_expired(self) -> bool:
    """Return whether the cache is passed its expiration date."""
    cutoff = DateTime.utcnow() - self.cache_ttl
    mtime = DateTime.utcfromtimestamp(os.path.getmtime(self.cache_file_path))
    return mtime < cutoff

_purge_icalendar_cache()

Purge the cache we have for this Calendar.

Source code in ical_library/cache_client.py
76
77
78
79
80
def _purge_icalendar_cache(self) -> None:
    """Purge the cache we have for this Calendar."""
    if self.verbose:
        print(f"Cache was expired. Removed {self.cache_file_path}.")
    return self.cache_file_path.unlink()

_write_response_to_cache(text)

Write the response of the fetched URL to cache.

Parameters:

Name Type Description Default
text str

The fetched result.

required
Source code in ical_library/cache_client.py
65
66
67
68
69
70
71
72
73
74
def _write_response_to_cache(self, text: str) -> None:
    """
    Write the response of the fetched URL to cache.
    :param text: The fetched result.
    """
    if self.verbose:
        print(f"Successfully loaded new iCalendar data and stored it at {self.cache_file_path}.")
    self.cache_file_path.parent.mkdir(parents=True, exist_ok=True)
    with open(self.cache_file_path, "w") as file:
        file.write(text)

get_icalendar(**kwargs)

Get a parsed VCalendar instance. If there is an active cache, return that, otherwise fetch and cache the result.

Parameters:

Name Type Description Default
kwargs Any

Any keyword arguments to pass onto the urllib.request.urlopen call.

{}

Returns:

Type Description
VCalendar

a VCalendar instance with all it's iCalendar components like VEvents, VToDos, VTimeZones etc.

Source code in ical_library/cache_client.py
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def get_icalendar(self, **kwargs: Any) -> VCalendar:
    """
    Get a parsed VCalendar instance. If there is an active cache, return that, otherwise fetch and cache the result.
    :param kwargs: Any keyword arguments to pass onto the `urllib.request.urlopen` call.
    :return: a VCalendar instance with all it's iCalendar components like VEvents, VToDos, VTimeZones etc.
    """
    if not self._is_cache_expired():
        if self.verbose:
            print("Using cache to remove this folder.")
        return client.parse_icalendar_file(self.cache_file_path)

    self._purge_icalendar_cache()
    response = request.urlopen(self.url, **kwargs)
    if not (200 <= response.getcode() < 400):
        raise ValueError(f"Unable to execute request at {self.url=}. Response code was: {response.getcode()}.")
    text = response.read().decode("utf-8")
    self._write_response_to_cache(text)

    lines = text.split("\n")
    return client.parse_lines_into_calendar(lines)