Coverage for src / anpr2mqtt / api_client.py: 96%

1import re 

2from pathlib import Path 

3from typing import TYPE_CHECKING, Any, Literal, cast 

4 

5import niquests 

6import structlog 

7from requests_cache.backends.filesystem import FileCache 

8from requests_cache.session import CacheMixin 

9 

10from anpr2mqtt.settings import CacheType 

11 

12if TYPE_CHECKING: 

13 from requests_cache.models.response import CachedResponse 

14 

15 

16class _CachedSession(CacheMixin, niquests.Session): # type: ignore[misc] 

17 """requests-cache backed by niquests as the transport.""" 

18 

19 

20log = structlog.get_logger() 

21 

22 

23class APIClient: 

24 def lookup(self, reg: str) -> dict[str, Any]: 

25 raise NotImplementedError() 

26 

27 

28class DVLAClient(APIClient): 

29 ID = "GB" 

30 REG_RE = r"(^[A-Z]{2}[0-9]{2}\s?[A-Z]{3}$)|(^[A-Z][0-9]{1,3}[A-Z]{3}$)|(^[A-Z]{3}[0-9]{1,3}[A-Z]$)|(^[0-9]{1,4}[A-Z]{1,2}$)|(^[0-9]{1,3}[A-Z]{1,3}$)|(^[A-Z]{1,2}[0-9]{1,4}$)|(^[A-Z]{1,3}[0-9]{1,3}$)|(^[A-Z]{1,3}[0-9]{1,4}$)|(^[0-9]{3}[DX]{1}[0-9]{3}$)" # noqa: E501 

31 """https://developer-portal.driver-vehicle-licensing.api.gov.uk""" 

32 

33 def __init__( 

34 self, 

35 api_key: str, 

36 cache_ttl: int = 60 * 60 * 6, 

37 cache_type: CacheType = CacheType.MEMORY, 

38 cache_dir: Path | None = None, 

39 verify_plate: str | None = None, 

40 test: bool = False, 

41 ) -> None: 

42 self.cache_session: _CachedSession | None = None 

43 if cache_type == CacheType.FILE and cache_dir: 

44 try: 

45 file_cache: FileCache = FileCache(cache_name=str(cache_dir), use_cache_dir=True) 

46 log.debug("Caching DVLA at %s for %s", file_cache.cache_dir, cache_ttl) 

47 self.cache_session = _CachedSession( 

48 cache_name="dvla_cache", allowable_methods=["GET", "POST"], expire_after=cache_ttl, backend=file_cache 

49 ) 

50 except Exception as e: 

51 log.error("Unable to configure file system caching, reverting to in memory: %s", e) 

52 cache_type = CacheType.MEMORY 

53 

54 if self.cache_session is None: 

55 if cache_type != CacheType.MEMORY: 

56 log.warn("Unable to cache DVLA as %s, falling back to in memory for %s", cache_type, cache_ttl) 

57 else: 

58 log.debug("Caching DVLA in memory for %s", cache_ttl) 

59 self.cache_session = _CachedSession( 

60 cache_name="dvla_cache", allowable_methods=["GET", "POST"], backend="memory", expire_after=cache_ttl 

61 ) 

62 

63 self.api_key: str = api_key 

64 self.env_prefix: Literal["uat."] | Literal[""] = "uat." if test else "" 

65 if verify_plate: 

66 result: dict[str, Any] = self.lookup(reg=verify_plate) 

67 if result and result["success"]: 

68 log.info("Verified DVLA API lookup using %s, details: %s", verify_plate, result["plate"]) 

69 else: 

70 log.error("DVLA startup verificatio failed: %s", result) 

71 

72 def lookup(self, reg: str) -> dict[str, Any]: 

73 if not re.match(self.REG_RE, reg): 

74 log.warning(f"DVLA SKIP invalid reg {reg}") 

75 return {"reg_match_fail": self.ID, "plate": {}, "success": False} 

76 if self.cache_session is None: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 log.error("Unable to lookup, failed to configure cache session or fallback") 

78 return {"lookup_fail": "missing cache", "plate": {}, "success": False} 

79 try: 

80 with self.cache_session as client: 

81 log.debug(f"Fetching DVLA info from API, cache_ttl={self.cache_session.expire_after}") 

82 response: CachedResponse = cast( 

83 "CachedResponse", 

84 client.post( 

85 url=f"https://{self.env_prefix}driver-vehicle-licensing.api.gov.uk/vehicle-enquiry/v1/vehicles", 

86 headers={"x-api-key": self.api_key, "Content-Type": "application/json"}, 

87 json={"registrationNumber": reg.upper()}, 

88 ), 

89 ) 

90 if response.from_cache: 

91 log.debug("DVLA API cached response, created %s", response.created_at) 

92 if response.status_code == 200: 

93 plate: dict[str, Any] = cast("dict[str,Any]", response.json()) 

94 return { 

95 "cache": { 

96 "calls": len(response.history) if response.history else 0, 

97 "cached": response.from_cache, 

98 "created": response.created_at.isoformat() if response.created_at else None, 

99 }, 

100 "plate": plate, 

101 "description": f"{plate.get('colour', '').title()} {plate.get('make', '').title()}" if plate else None, 

102 "success": True, 

103 } 

104 

105 log.error("DVLA API FAIL: %s", response.json()) 

106 return { 

107 "api_errors": response.json()["errors"], 

108 "api_status": response.status_code, 

109 "plate": {}, 

110 "success": False, 

111 } 

112 except Exception as e: 

113 log.exception("Failed to fetch DVLA reg data") 

114 return {"api_exception": str(e), "plate": {}, "success": False}