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( 

21 descr=i18n.translate( 

22 "viur.core.bones.address.street_name.descr", 

23 defaultText="Street", 

24 ), 

25 required=True, 

26 ) 

27 street_number = StringBone( 

28 descr=i18n.translate( 

29 "viur.core.bones.address.street_number.descr", 

30 defaultText="House number", 

31 ), 

32 ) 

33 address_addition = StringBone( 

34 descr=i18n.translate( 

35 "viur.core.bones.address.address_addition.descr", 

36 defaultText="Address addition", 

37 ), 

38 ) 

39 zip_code = StringBone( 

40 descr=i18n.translate( 

41 "viur.core.bones.address.zip_code.descr", 

42 defaultText="ZIP / Postal code", 

43 ), 

44 params={ 

45 "pattern": ZIP_CODE_PATTERNS, 

46 "pattern-error": i18n.translate( 

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

48 defaultText="Invalid ZIP code", 

49 ), 

50 }, 

51 ) 

52 city = StringBone( 

53 descr=i18n.translate( 

54 "viur.core.bones.address.city.descr", 

55 defaultText="City", 

56 ), 

57 required=True, 

58 ) 

59 state = StringBone( 

60 descr=i18n.translate( 

61 "viur.core.bones.address.state.descr", 

62 defaultText="State", 

63 ), 

64 ) 

65 country = SelectCountryBone( 

66 descr=i18n.translate( 

67 "viur.core.bones.address.country.descr", 

68 defaultText="Country", 

69 ), 

70 ) 

71 coordinates = SpatialBone( 

72 descr=i18n.translate( 

73 "viur.core.bones.address.coordinates.descr", 

74 defaultText="Coordinates", 

75 ), 

76 boundsLat=(-90, 90), 

77 boundsLng=(-180, 180), 

78 gridDimensions=(10, 10), 

79 ) 

80 

81 

82class AddressBone(RecordBone): 

83 type = RecordBone.type + ".address" 

84 

85 def __init__( 

86 self, 

87 *, 

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

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

90 **kwargs, 

91 ): 

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

93 

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

95 value = skel[name] 

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

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

98 coords = self.geocode(value) 

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

100 value["coordinates"] = coords 

101 

102 @staticmethod 

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

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

105 return db.Key(CACHE_KIND, digest) 

106 

107 @staticmethod 

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

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

110 params = { 

111 "street": street, 

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

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

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

115 "format": "json", 

116 "limit": 1, 

117 } 

118 

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

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

121 params["state"] = state 

122 

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

124 try: 

125 cached = db.get(cache_key) 

126 if cached is not None: 

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

128 response = requests.get( 

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

130 params=params, 

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

132 timeout=5, 

133 ) 

134 if response.status_code != 200: 

135 logging.error( 

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

137 ) 

138 return None 

139 data = response.json() 

140 if not data: 

141 return None 

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

143 entity = db.Entity(cache_key) 

144 entity["lat"] = lat 

145 entity["lng"] = lng 

146 db.put(entity) 

147 return lat, lng 

148 except Exception as e: 

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

150 return None