Browse Source

implement disk and runtime caching for PyPI package classifiers

master
Niklas Rosenstein 6 months ago
parent
commit
0aac86dc95
Signed by: NiklasRosenstein <rosensteinniklas@gmail.com> GPG Key ID: 06D269B33D25F6C6
1 changed files with 43 additions and 1 deletions
  1. +43
    -1
      src/shore/util/classifiers.py

+ 43
- 1
src/shore/util/classifiers.py View File

@@ -19,11 +19,53 @@
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.

from datetime import datetime
from typing import List
import logging
import os
import requests
import time

CACHE_FILENAME = os.path.expanduser('~/.local/shore/classifiers-cache.txt')
CACHE_TTL = 60 * 60 * 24 * 7 # 7 days
CLASSIFIERS_URL = 'https://pypi.org/pypi?%3Aaction=list_classifiers'
logger = logging.getLogger(__name__)
_runtime_cache = None


def get_classifiers() -> List[str]:
return requests.get(CLASSIFIERS_URL).text.split('\n')
global _runtime_cache
if _runtime_cache is not None:
return list(_runtime_cache)

def _load_cachefile():
global _runtime_cache
with open(CACHE_FILENAME) as fp:
_runtime_cache = [x.rstrip('\n') for x in fp]
return list(_runtime_cache)

has_cachefile = os.path.isfile(CACHE_FILENAME)
if has_cachefile and (time.time() - os.path.getmtime(CACHE_FILENAME)) < CACHE_TTL:
return _load_cachefile()
try:
classifiers = requests.get(CLASSIFIERS_URL, timeout=1).text.split('\n')
except requests.exceptions.ReadTimeout as exc:
if has_cachefile:
cache_timestamp = datetime.fromtimestamp(os.path.getmtime(CACHE_FILENAME))
message = ' Falling back to cached classifiers (timestamp: {})'.format(cache_timestamp)
else:
message = ' Returning no classifiers. Subsequent classifier validation might raise.'
logger.warning('Error retrieving classifiers from "%s" (%s).%s', CLASSIFIERS_URL, exc, message)
if has_cachefile:
return _load_cachefile()
_runtime_cache = []
return []
try:
os.makedirs(os.path.dirname(CACHE_FILENAME), exist_ok=True)
with open(CACHE_FILENAME, 'w') as fp:
fp.writelines(classifiers)
except:
logger.exception('Unable to write classifiers cache file.')

_runtime_cache = classifiers
return list(classifiers)

Loading…
Cancel
Save