Coverage for an_website / quotes / quote_of_the_day / store.py: 85.294%

1# This program is free software: you can redistribute it and/or modify 

2# it under the terms of the GNU Affero General Public License as 

3# published by the Free Software Foundation, either version 3 of the 

4# License, or (at your option) any later version. 

5# 

6# This program is distributed in the hope that it will be useful, 

7# but WITHOUT ANY WARRANTY; without even the implied warranty of 

8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

9# GNU Affero General Public License for more details. 

10# 

11# You should have received a copy of the GNU Affero General Public License 

12# along with this program. If not, see <https://www.gnu.org/licenses/>. 

13 

14"""Stores that contain the ids of old quotes of the day.""" 

15 

16 

17import abc 

18from datetime import date, datetime, timezone 

19from typing import ClassVar, Final 

20 

21from redis.asyncio import Redis 

22from tornado.web import HTTPError 

23 

24from ... import EVENT_REDIS 

25 

26QUOTE_COUNT_TO_SHOW_IN_FEED: Final[int] = 8 

27 

28 

29class QuoteOfTheDayStore(abc.ABC): 

30 """The class representing the store for the quote of the day.""" 

31 

32 __slots__ = () 

33 

34 @abc.abstractmethod 

35 async def get_quote_id_by_date(self, date_: date) -> tuple[int, int] | None: 

36 """Get the quote id for the given date.""" 

37 raise NotImplementedError 

38 

39 @abc.abstractmethod 

40 async def has_quote_been_used(self, quote_id: tuple[int, int]) -> bool: 

41 """Check if the quote has been used already.""" 

42 raise NotImplementedError 

43 

44 @abc.abstractmethod 

45 async def set_quote_id_by_date( 

46 self, date_: date, quote_id: tuple[int, int] 

47 ) -> None: 

48 """Set the quote id for the given date.""" 

49 raise NotImplementedError 

50 

51 @abc.abstractmethod 

52 async def set_quote_to_used(self, quote_id: tuple[int, int]) -> None: 

53 """Set the quote as used.""" 

54 raise NotImplementedError 

55 

56 

57class QuoteOfTheDayStoreWithCache(QuoteOfTheDayStore, abc.ABC): 

58 """Quote of the day store with an in memory cache.""" 

59 

60 # pylint: disable=abstract-method 

61 

62 CACHE: ClassVar[dict[date, tuple[int, int]]] 

63 

64 @classmethod 

65 def _get_quote_id_from_cache(cls, date_: date) -> None | tuple[int, int]: 

66 """Get a quote_id from the cache if it is present.""" 

67 return cls.CACHE.get(date_) 

68 

69 @classmethod 

70 def _populate_cache(cls, date_: date, quote_id: tuple[int, int]) -> None: 

71 """Populate the cache for the quote of today.""" 

72 today = datetime.now(timezone.utc).date() 

73 # old entries are rarely used, they don't need to be cached 

74 if (today - date_).days > QUOTE_COUNT_TO_SHOW_IN_FEED: 

75 cls.CACHE[date_] = quote_id 

76 

77 for key in tuple(cls.CACHE): 

78 # remove old entries from cache to save memory 

79 if (today - key).days > QUOTE_COUNT_TO_SHOW_IN_FEED: 

80 del cls.CACHE[key] 

81 

82 

83class RedisQuoteOfTheDayStore(QuoteOfTheDayStoreWithCache): 

84 """A quote of the day store that stores the quote of the day in Redis.""" 

85 

86 __slots__ = ("redis", "redis_prefix") 

87 

88 CACHE: ClassVar[dict[date, tuple[int, int]]] = {} 

89 

90 redis_prefix: str 

91 redis: Redis[str] 

92 

93 def __init__(self, redis: Redis[str], redis_prefix: str) -> None: 

94 """Initialize the Redis quote of the day store.""" 

95 super().__init__() 

96 self.redis = redis 

97 self.redis_prefix = redis_prefix 

98 

99 async def get_quote_id_by_date(self, date_: date) -> tuple[int, int] | None: 

100 """Get the quote id for the given date.""" 

101 if date_ in self.CACHE: 

102 return self.CACHE[date_] 

103 if not EVENT_REDIS.is_set(): 

104 raise HTTPError(503) 

105 wq_id = await self.redis.get(self.get_redis_quote_date_key(date_)) 

106 if not wq_id: 

107 return None 

108 quote, author = wq_id.split("-") 

109 quote_id = int(quote), int(author) 

110 self._populate_cache(date_, quote_id) 

111 return quote_id 

112 

113 def get_redis_quote_date_key(self, wq_date: date) -> str: 

114 """Get the Redis key for getting quotes by date.""" 

115 return f"{self.redis_prefix}:quote-of-the-day:by-date:{wq_date.isoformat()}" 

116 

117 def get_redis_used_key(self, wq_id: tuple[int, int]) -> str: 

118 """Get the Redis used key.""" 

119 str_id = "-".join(map(str, wq_id)) # pylint: disable=bad-builtin 

120 return f"{self.redis_prefix}:quote-of-the-day:used:{str_id}" 

121 

122 async def has_quote_been_used(self, quote_id: tuple[int, int]) -> bool: 

123 """Check if the quote has been used already.""" 

124 if quote_id in self.CACHE.values(): 

125 return True 

126 if not EVENT_REDIS.is_set(): 

127 raise HTTPError(503) 

128 return bool(await self.redis.get(self.get_redis_used_key(quote_id))) 

129 

130 async def set_quote_id_by_date( 

131 self, date_: date, quote_id: tuple[int, int] 

132 ) -> None: 

133 """Set the quote id for the given date.""" 

134 if not EVENT_REDIS.is_set(): 

135 raise HTTPError(503) 

136 await self.redis.setex( 

137 self.get_redis_quote_date_key(date_), 

138 60 * 60 * 24 * 420, # TTL 

139 "-".join(map(str, quote_id)), # pylint: disable=bad-builtin 

140 ) 

141 self._populate_cache(date_, quote_id) 

142 

143 async def set_quote_to_used(self, quote_id: tuple[int, int]) -> None: 

144 """Set the quote as used.""" 

145 if not EVENT_REDIS.is_set(): 

146 return 

147 

148 await self.redis.setex( 

149 self.get_redis_used_key(quote_id), 

150 # we have over 720 funny wrong quotes, so 420 should be ok 

151 60 * 60 * 24 * 420, # TTL 

152 1, # True 

153 )