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

1import re 

2from typing import TYPE_CHECKING, Any, cast 

3 

4import structlog 

5from hishel.httpx import SyncCacheClient 

6 

7if TYPE_CHECKING: 

8 from httpx import Response 

9 

10log = structlog.get_logger() 

11 

12 

13class APIClient: 

14 def lookup(self, reg: str) -> list[Any] | dict[str, Any] | None: 

15 raise NotImplementedError() 

16 

17 

18class DVLA(APIClient): 

19 ID = "GB" 

20 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 

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

22 

23 def __init__(self, api_key: str, cache_ttl: int = 60 * 60 * 6) -> None: 

24 self.cache_ttl: int = cache_ttl 

25 self.api_key: str = api_key 

26 

27 def lookup(self, reg: str) -> list[Any] | dict[str, Any] | None: 

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

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

30 return {"reg_match_fail": self.ID} 

31 try: 

32 with SyncCacheClient(headers=[("cache-control", f"max-age={self.cache_ttl}")]) as client: 

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

34 response: Response = client.post( 

35 url="https://driver-vehicle-licensing.api.gov.uk/vehicle-enquiry/v1/vehicles", 

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

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

38 ) 

39 if response.extensions.get("hishel_from_cache"): 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true

40 log.debug("DVLA API cached response") 

41 if response.status_code == 200: 

42 return cast("dict[str,Any]", response.json()) 

43 

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

45 return {"api_errors": response.json()["errors"], "api_status": response.status_code} 

46 except Exception as e: 

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

48 return {"api_exception": str(e)}