Coverage for src\baobab_web_api_caller\config\service_config.py: 86%

1"""Configuration d'un service distant.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from types import MappingProxyType 

7from typing import Mapping 

8from urllib.parse import urlparse 

9 

10from baobab_web_api_caller.auth.authentication_strategy import AuthenticationStrategy 

11from baobab_web_api_caller.auth.no_authentication_strategy import NoAuthenticationStrategy 

12from baobab_web_api_caller.config.rate_limit_policy import RateLimitPolicy 

13from baobab_web_api_caller.config.retry_policy import RetryPolicy 

14from baobab_web_api_caller.exceptions.configuration_exception import ConfigurationException 

15from baobab_web_api_caller.utils.mapping_utils import freeze_str_mapping 

16 

17 

18@dataclass(frozen=True, slots=True) 

19class ServiceConfig: 

20 """Configuration transverse d'un service distant. 

21 

22 :param base_url: Base URL du service (ex: ``https://api.example.com``). 

23 :type base_url: str 

24 :param default_headers: Headers appliqués par défaut à chaque requête. 

25 :type default_headers: Mapping[str, str] 

26 :param authentication_strategy: Stratégie d'authentification à appliquer. 

27 :type authentication_strategy: AuthenticationStrategy 

28 :param default_timeout_seconds: Timeout par défaut en secondes. 

29 :type default_timeout_seconds: float | None 

30 :param retry_policy: Politique de retry. 

31 :type retry_policy: RetryPolicy 

32 :param rate_limit_policy: Politique de throttling. 

33 :type rate_limit_policy: RateLimitPolicy 

34 :raises ConfigurationException: Si la configuration est invalide. 

35 """ 

36 

37 base_url: str 

38 default_headers: Mapping[str, str] = MappingProxyType({}) 

39 authentication_strategy: AuthenticationStrategy = NoAuthenticationStrategy() 

40 default_timeout_seconds: float | None = None 

41 retry_policy: RetryPolicy = RetryPolicy() 

42 rate_limit_policy: RateLimitPolicy = RateLimitPolicy() 

43 

44 def __post_init__(self) -> None: 

45 normalized = self._normalize_and_validate_base_url(self.base_url) 

46 object.__setattr__(self, "base_url", normalized) 

47 object.__setattr__( 

48 self, "default_headers", freeze_str_mapping(self.default_headers, "default_headers") 

49 ) 

50 

51 if not isinstance(self.authentication_strategy, AuthenticationStrategy): 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true

52 raise ConfigurationException( 

53 "authentication_strategy must be an AuthenticationStrategy" 

54 ) 

55 

56 if self.default_timeout_seconds is not None and self.default_timeout_seconds <= 0: 

57 raise ConfigurationException("default_timeout_seconds must be positive when provided") 

58 

59 if not isinstance(self.retry_policy, RetryPolicy): 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

60 raise ConfigurationException("retry_policy must be a RetryPolicy") 

61 if not isinstance(self.rate_limit_policy, RateLimitPolicy): 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true

62 raise ConfigurationException("rate_limit_policy must be a RateLimitPolicy") 

63 

64 @staticmethod 

65 def _normalize_and_validate_base_url(base_url: str) -> str: 

66 if not isinstance(base_url, str) or base_url.strip() == "": 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 raise ConfigurationException("base_url must be a non-empty string") 

68 

69 trimmed = base_url.strip() 

70 parsed = urlparse(trimmed) 

71 if parsed.scheme not in {"http", "https"}: 

72 raise ConfigurationException("base_url must start with http:// or https://") 

73 if parsed.netloc == "": 

74 raise ConfigurationException("base_url must include a host") 

75 

76 normalized = trimmed.rstrip("/") 

77 return normalized