Coverage for src\baobab_web_api_caller\transport\throttler.py: 97%

1"""Throttling (intervalle minimal) testable.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6 

7from baobab_web_api_caller.config.rate_limit_policy import RateLimitPolicy 

8from baobab_web_api_caller.transport.sleeper import Sleeper 

9from baobab_web_api_caller.transport.time_provider import TimeProvider 

10 

11 

12@dataclass(slots=True) 

13class Throttler: 

14 """Applique une politique de limitation de débit.""" 

15 

16 rate_limit_policy: RateLimitPolicy 

17 time_provider: TimeProvider 

18 sleeper: Sleeper 

19 _last_call_ts: float | None = None 

20 

21 def throttle(self) -> None: 

22 """Attend si nécessaire pour respecter l'intervalle minimal.""" 

23 

24 min_interval = float(self.rate_limit_policy.min_interval_seconds) 

25 if min_interval <= 0: 

26 return 

27 

28 now = float(self.time_provider.monotonic()) 

29 if self._last_call_ts is None: 

30 self._last_call_ts = now 

31 return 

32 

33 elapsed = now - self._last_call_ts 

34 remaining = min_interval - elapsed 

35 if remaining > 0: 35 ↛ 39line 35 didn't jump to line 39 because the condition on line 35 was always true

36 self.sleeper.sleep(remaining) 

37 now = float(self.time_provider.monotonic()) 

38 

39 self._last_call_ts = now