|
1 | 1 | import logging |
2 | 2 | from abc import ABC, abstractmethod |
3 | 3 | from datetime import datetime |
4 | | -from typing import Any |
| 4 | +from typing import Any, Union |
5 | 5 |
|
6 | 6 | import requests |
7 | 7 |
|
|
19 | 19 | class AbstractPackageReference(ABC): |
20 | 20 | """Represents a reference from where to retrieve trusted packages.""" |
21 | 21 |
|
22 | | - def __init__(self, source: str, cache_handler: CacheHandler) -> None: |
| 22 | + def __init__(self, source: str, cache_handler: Union[CacheHandler, None] = None) -> None: |
23 | 23 | self.source = source |
24 | 24 | self.cache_handler = cache_handler |
25 | 25 |
|
26 | 26 | @abstractmethod |
27 | | - def get_packages(self, use_cache: bool = True) -> set[str]: |
| 27 | + def get_packages(self) -> set[str]: |
28 | 28 | """Return the names of the trusted packages available in the reference.""" |
29 | 29 |
|
30 | 30 |
|
31 | 31 | class TopPyPiReference(AbstractPackageReference): |
32 | 32 | """Top PyPi packages retrieved from an online source.""" |
33 | 33 |
|
34 | | - def get_packages(self, use_cache: bool = True) -> set[str]: |
| 34 | + def get_packages(self) -> set[str]: |
35 | 35 | """Download and parse online source of top Python Package Index packages.""" |
36 | 36 | packages_to_use = set() |
37 | | - if use_cache: |
38 | | - packages_to_use = self._get_packages_from_cache() |
39 | | - # we don't save the cache here, we keep it as it is so the date remains the original one. |
| 37 | + packages_to_use = self._get_packages_from_cache_if_enabled() |
| 38 | + # we don't save the cache here, we keep it as it is so the date remains the original one. |
40 | 39 |
|
41 | 40 | if not packages_to_use: |
42 | 41 | # no cache usage, no cache hit (non-existent or outdated) or cache was empty. |
43 | 42 | logger.info("Fetching trusted packages from PyPI reference...") |
44 | 43 | packages_to_use = self._parse(self._download()) |
45 | | - if use_cache: |
46 | | - self._save_trusted_packages_to_cache(packages_to_use) |
| 44 | + |
| 45 | + # New packages were downloaded, we create a new entry updating all values. |
| 46 | + self._save_trusted_packages_to_cache_if_enabled(packages_to_use) |
47 | 47 |
|
48 | 48 | normalized_packages = normalize_packages(packages_to_use) |
49 | 49 | return normalized_packages |
50 | 50 |
|
51 | | - def _save_trusted_packages_to_cache(self, packages: set[str]) -> None: |
| 51 | + def _save_trusted_packages_to_cache_if_enabled(self, packages: set[str]) -> None: |
52 | 52 | """Save trusted packages using CacheHandler.""" |
| 53 | + if not self.cache_handler: |
| 54 | + return |
53 | 55 | cache_entry = CacheEntry(saved_date=datetime.now().date().isoformat(), packages=packages) |
54 | 56 | self.cache_handler.write_entry(self.source, cache_entry) |
55 | 57 | logger.debug("Saved %d trusted packages for source %s", len(packages), self.source) |
56 | 58 |
|
57 | | - def _get_packages_from_cache(self) -> set[str]: |
| 59 | + def _get_packages_from_cache_if_enabled(self) -> set[str]: |
58 | 60 | """Get packages from cache if it's present and up to date.""" |
| 61 | + if not self.cache_handler: |
| 62 | + return set() |
59 | 63 | cache_entry = self.cache_handler.get_cache_entry(self.source) |
60 | 64 | if not cache_entry: |
61 | 65 | logger.debug("No cache entry found for source: %s", self.source) |
|
0 commit comments