Coverage for r11k/httpcache.py: 59%

1"""Extension of `r11k.cache.KeyValueStore`. Should possibly be merged.""" 

2 

3import os 

4import os.path 

5from datetime import datetime 

6import time 

7import requests 

8from pathlib import Path 

9 

10from typing import Any, Optional 

11 

12import r11k 

13from r11k.cache import KeyValueStore 

14 

15 

16class HTTPCache: 

17 """ 

18 HTTP client with built in cache. 

19 

20 First caches any (successful) request for one hour (configurable), 

21 after the local TTL has expired the request is re-fired, but with 

22 a 'If-Modified-Since' header set. 

23 

24 :param path: Where in the filesystem cached files are stored. 

25 :param default_ttl: How long (in seconds) we should cache files before re-fetching. 

26 """ 

27 

28 def __init__(self, path: str = '/tmp', default_ttl: int = 3600): 

29 self.store = KeyValueStore(path) 

30 self.default_ttl = default_ttl 

31 

32 def get(self, url: str) -> Optional[Any]: 

33 """ 

34 Fetch (from internet or cache) the contents of url. 

35 

36 :return: A json blob 

37 """ 

38 key = url 

39 if existing := self.store.get(key): 

40 path = self.store.path(key) 

41 try: 

42 st = os.stat(path) 

43 last_modified = datetime.utcfromtimestamp(st.st_mtime) 

44 now = datetime.utcfromtimestamp(time.time()) 

45 if abs(now - last_modified).seconds < self.default_ttl: 

46 return existing 

47 except FileNotFoundError: 

48 return existing 

49 

50 response = self.__fetch(url, last_modified) 

51 if response.status_code == 304: 

52 Path(path).touch() 

53 return existing 

54 elif response.status_code != 200: 

55 raise RuntimeError('Invalid response code', response.status_code, url) 

56 

57 data = response.json() 

58 self.store.put(key, data) 

59 return data 

60 

61 else: 

62 response = self.__fetch(url) 

63 if response.status_code != 200: 63 ↛ 64line 63 didn't jump to line 64, because the condition on line 63 was never true

64 raise RuntimeError('Invalid response code', response.status_code, url) 

65 data = response.json() 

66 self.store.put(key, data) 

67 return data 

68 

69 def __fetch(self, url: str, last_modified: Optional[datetime] = None) -> requests.Response: 

70 headers = {'User-Agent': f'r11k/{r11k.VERSION}'} 

71 if last_modified: 71 ↛ 72line 71 didn't jump to line 72, because the condition on line 71 was never true

72 date = last_modified.strftime('%a, %d %b %Y %H:%M:%S GMT') 

73 headers['If-Modified-Since'] = date 

74 return requests.get(url, headers=headers)