Coverage for packages/python/voice_prices/update_prices.py: 91.96%

1from __future__ import annotations as _annotations (empty)

2 

3import asyncio (empty)

4import logging (empty)

5import threading (empty)

6from dataclasses import dataclass, field (empty)

7from time import time (empty)

8 

9import httpx2 (empty)

10 

11from . import data_snapshot (empty)

12 

13__all__ = ( (empty)

14 'DEFAULT_UPDATE_URL', 

15 'UpdatePrices', 

16 'wait_prices_updated_sync', 

17 'wait_prices_updated_async', 

18) 

19 

20logger = logging.getLogger('voice-prices') (empty)

21DEFAULT_UPDATE_URL = 'https://raw.githubusercontent.com/mahimailabs/voice-prices/refs/heads/main/prices/data.json' (empty)

22_global_update_prices: UpdatePrices | None = None (empty)

23 

24 

25def wait_prices_updated_sync(timeout: float | None = None) -> bool: (empty)

26 """Synchronously wait for prices to be updated. 

27 

28 Args: 

29 timeout: The maximum time to wait for prices to be updated. Defaults to None which waits indefinitely. 

30 

31 Returns: 

32 True if prices were updated, False otherwise. 

33 """ 

34 if _global_update_prices: 34 ↛ 36line 34 didn't jump to line 36 because the condition on line 34 was always true(empty)

35 return _global_update_prices.wait(timeout) (empty)

36 return False 

37 

38 

39async def wait_prices_updated_async(timeout: float | None = None) -> bool: (empty)

40 """Asynchronously wait for prices to be updated. 

41 

42 Args: 

43 timeout: The maximum time to wait for prices to be updated. Defaults to None which waits indefinitely. 

44 

45 Returns: 

46 True if prices were updated, False otherwise. 

47 """ 

48 return await asyncio.to_thread(wait_prices_updated_sync, timeout) (empty)

49 

50 

51@dataclass (empty)

52class UpdatePrices: (empty)

53 """Update prices in the background using a daemon thread. 

54 

55 Can be used either as a context manager or as a simple class, where you'll need to call start() and stop() manually. 

56 """ 

57 

58 update_interval: float = 3600 (empty)

59 """How often to update prices in seconds.""" (empty)

60 url: str = DEFAULT_UPDATE_URL (empty)

61 """The URL to fetch prices from.""" (empty)

62 request_timeout: httpx2.Timeout = field(default_factory=lambda: httpx2.Timeout(timeout=10, connect=5)) (empty)

63 """The timeout for HTTP requests.""" (empty)

64 _stop_event: threading.Event = field(default_factory=threading.Event) (empty)

65 _prices_updated: threading.Event = field(default_factory=threading.Event) (empty)

66 _thread: threading.Thread | None = field(default=None, init=False) (empty)

67 _background_exc: Exception | None = field(default=None, init=False) (empty)

68 

69 def start(self, *, wait: bool | float = False): (empty)

70 """Start the background task. 

71 

72 Args: 

73 wait: Whether to wait for the prices to be updated before returning, if an int is passed 

74 wait for that many seconds, if `True` wait for 30 seconds. 

75 """ 

76 global _global_update_prices 

77 

78 if self._thread is not None: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true(empty)

79 raise RuntimeError('UpdatePrices background task already started') 

80 

81 if _global_update_prices is not None: (empty)

82 raise RuntimeError( (empty)

83 'UpdatePrices global task already started, only one UpdatePrices can be active at a time' 

84 ) 

85 

86 _global_update_prices = self (empty)

87 self._prices_updated.clear() (empty)

88 self._stop_event.clear() (empty)

89 self._background_exc = None (empty)

90 self._thread = threading.Thread(target=self._background_task, daemon=True, name='voice_prices:update') (empty)

91 self._thread.start() (empty)

92 if wait: 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true(empty)

93 self.wait(timeout=30 if wait is True else wait) 

94 

95 def wait(self, timeout: float | None = None) -> bool: (empty)

96 """Wait for the prices to be updated in the background task. 

97 

98 Args: 

99 timeout: The maximum time to wait for the prices to be updated in seconds. 

100 """ 

101 prices_updated = self._prices_updated.wait(timeout=timeout) (empty)

102 exc = self._background_exc (empty)

103 if exc: (empty)

104 self._background_exc = None (empty)

105 raise exc (empty)

106 return prices_updated (empty)

107 

108 def stop(self): (empty)

109 """Stop the background task.""" 

110 global _global_update_prices 

111 

112 _global_update_prices = None (empty)

113 if self._thread is not None: (empty)

114 self._stop_event.set() (empty)

115 self._thread.join() (empty)

116 self._thread = None (empty)

117 # Clear after the thread exits so an in-flight fetch cannot reinstall a snapshot after stop(). 

118 data_snapshot.set_custom_snapshot(None) (empty)

119 if self._background_exc: (empty)

120 exc = self._background_exc (empty)

121 self._background_exc = None (empty)

122 raise exc (empty)

123 

124 def __enter__(self): (empty)

125 self.start() (empty)

126 return self (empty)

127 

128 def __exit__(self, *_args: object): (empty)

129 self.stop() (empty)

130 

131 def _background_task(self) -> None: (empty)

132 logger.info('Starting voice-prices background task') (empty)

133 try: (empty)

134 while True: (empty)

135 try: (empty)

136 self._update_prices() (empty)

137 self._prices_updated.set() (empty)

138 self._background_exc = None (empty)

139 except Exception as e: (empty)

140 self._background_exc = e (empty)

141 self._prices_updated.set() (empty)

142 logger.error('Error updating voice-prices in the background (%s): %s', type(e).__name__, e) (empty)

143 if self._stop_event.wait(self.update_interval): 143 ↛ 134line 143 didn't jump to line 134 because the condition on line 143 was always true(empty)

144 break (empty)

145 

146 finally: 

147 logger.info('voice-prices background task stopped') (empty)

148 

149 def _update_prices(self): (empty)

150 start = time() (empty)

151 snapshot = self.fetch() (empty)

152 interval = time() - start (empty)

153 if snapshot: 153 ↛ 156line 153 didn't jump to line 156 because the condition on line 153 was always true(empty)

154 logger.info('Successfully fetched %d providers in %.2f seconds', len(snapshot.providers), interval) (empty)

155 else: 

156 logger.info('Successfully fetched null snapshot in %.2f seconds', interval) 

157 

158 data_snapshot.set_custom_snapshot(snapshot) (empty)

159 

160 def fetch(self) -> data_snapshot.DataSnapshot | None: (empty)

161 """Fetches the latest provider data from the configured URL.""" 

162 from . import data (empty)

163 

164 r = httpx2.get(self.url, timeout=self.request_timeout) (empty)

165 r.raise_for_status() (empty)

166 return data_snapshot.DataSnapshot(data.providers_schema.validate_json(r.content), from_auto_update=True) (empty)