Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/bones/address.py: 99%

1import hashlib 

2import logging 

3import typing as t 

4import urllib.parse 

5 

6import requests 

7 

8from .record import RecordBone 

9from .string import StringBone 

10from .selectcountry import SelectCountryBone 

11from .spatial import SpatialBone 

12from ..skeleton.relskel import RelSkel 

13from .. import db, i18n 

14from ..data.zip_patterns import ZIP_CODE_PATTERNS 

15 

16CACHE_KIND = "viur-addressbone-geocache" 

17 

18 

19class AddressRelSkel(RelSkel): 

20 street_name = StringBone(descr="Street", required=True) 

21 street_number = StringBone(descr="House number") 

22 address_addition = StringBone(descr="Address addition") 

23 zip_code = StringBone( 

24 descr="ZIP / Postal code", 

25 params={ 

26 "pattern": ZIP_CODE_PATTERNS, 

27 "pattern-error": i18n.translate( 

28 "viur.core.bones.address.zip_code.invalid", 

29 defaultText="Invalid ZIP code", 

30 ), 

31 }, 

32 ) 

33 city = StringBone(descr="City", required=True) 

34 state = StringBone(descr="State") 

35 country = SelectCountryBone(descr="Country") 

36 coordinates = SpatialBone( 

37 descr="Coordinates", 

38 boundsLat=(-90, 90), 

39 boundsLng=(-180, 180), 

40 gridDimensions=(10, 10), 

41 ) 

42 

43 

44class AddressBone(RecordBone): 

45 type = RecordBone.type + ".address" 

46 

47 def __init__( 

48 self, 

49 *, 

50 using: t.Type[RelSkel] = AddressRelSkel, 

51 format: str = "$(street_name) $(street_number), $(zip_code) $(city)", 

52 **kwargs, 

53 ): 

54 super().__init__(using=using, format=format, **kwargs) 

55 

56 def after_from_client(self, skel, name, errors): 

57 value = skel[name] 

58 # Coordinates sent by the client (e.g. a pin set on a map) are more precise than geocoding 

59 if value is not None and self.using.coordinates.isEmpty(value["coordinates"]): 

60 coords = self.geocode(value) 

61 if coords: 61 ↛ exitline 61 didn't return from function 'after_from_client' because the condition on line 61 was always true

62 value["coordinates"] = coords 

63 

64 @staticmethod 

65 def _cache_key(params: str) -> db.Key: 

66 digest = hashlib.sha256(params.encode()).hexdigest() 

67 return db.Key(CACHE_KIND, digest) 

68 

69 @staticmethod 

70 def geocode(skel: RelSkel) -> tuple[float, float] | None: 

71 street = f"{skel['street_name'] or ''} {skel['street_number'] or ''}".strip() 

72 params = { 

73 "street": street, 

74 "postalcode": skel["zip_code"] or "", 

75 "city": skel["city"] or "", 

76 "country": skel["country"] or "", 

77 "format": "json", 

78 "limit": 1, 

79 } 

80 

81 # Only when set, so cache keys of addresses without a state stay the same; `using` may lack the field 

82 if state := skel.get("state"): 

83 params["state"] = state 

84 

85 cache_key = AddressBone._cache_key(urllib.parse.urlencode(params)) 

86 try: 

87 cached = db.get(cache_key) 

88 if cached is not None: 

89 return cached["lat"], cached["lng"] 

90 response = requests.get( 

91 "https://nominatim.openstreetmap.org/search", 

92 params=params, 

93 headers={"User-Agent": "viur-addressbone/1.0 (viur.dev)"}, 

94 timeout=5, 

95 ) 

96 if response.status_code != 200: 

97 logging.error( 

98 f"AddressBone: Nominatim returned {response.status_code=}" 

99 ) 

100 return None 

101 data = response.json() 

102 if not data: 

103 return None 

104 lat, lng = float(data[0]["lat"]), float(data[0]["lon"]) 

105 entity = db.Entity(cache_key) 

106 entity["lat"] = lat 

107 entity["lng"] = lng 

108 db.put(entity) 

109 return lat, lng 

110 except Exception as e: 

111 logging.error(f"AddressBone: Nominatim geocoding failed with {e=}") 

112 return None