Coverage for gws-app/gws/base/ows/client/provider.py: 0%

1from typing import Optional 

2import base64 

3 

4import gws 

5import gws.gis.crs 

6import gws.gis.source 

7import gws.lib.net 

8import gws.lib.mime 

9 

10from . import request 

11 

12_IMAGE_VERBS = { 

13 gws.OwsVerb.GetLegendGraphic, 

14 gws.OwsVerb.GetMap, 

15 gws.OwsVerb.GetTile, 

16} 

17 

18_PREFER_IMAGE_MIME = [gws.lib.mime.PNG, gws.lib.mime.JPEG] 

19_PREFER_XML_MIME = [gws.lib.mime.GML3, gws.lib.mime.GML, gws.lib.mime.XML] 

20 

21 

22class OperationConfig(gws.Config): 

23 """Custom OWS operation.""" 

24 verb: gws.OwsVerb 

25 """OWS verb.""" 

26 formats: Optional[list[str]] 

27 """Supported formats.""" 

28 params: Optional[dict] 

29 """Operation parameters. (added in 8.1)""" 

30 postUrl: Optional[gws.Url] 

31 """URL for POST requests.""" 

32 url: Optional[gws.Url] 

33 """URL for GET requests.""" 

34 

35 

36class AuthorizationConfig(gws.Config): 

37 """Service authorization. (added in 8.1)""" 

38 type: str 

39 """Authorization type (only "basic" is supported).""" 

40 username: str = '' 

41 """User name.""" 

42 password: str = '' 

43 """Password.""" 

44 

45 

46class Config(gws.Config): 

47 """OWS provider configuration.""" 

48 

49 capsCacheMaxAge: gws.Duration = '1d' 

50 """Max cache age for capabilities documents.""" 

51 forceCrs: Optional[gws.CrsName] 

52 """Use this CRS for requests.""" 

53 alwaysXY: bool = False 

54 """Force XY orientation for lat/lon projections.""" 

55 maxRequests: int = 0 

56 """Max concurrent requests to this source.""" 

57 operations: Optional[list[OperationConfig]] 

58 """Override operations reported in capabilities.""" 

59 authorization: Optional[AuthorizationConfig] 

60 """Service authorization. (added in 8.1)""" 

61 url: gws.Url 

62 """Service url.""" 

63 

64 

65class Object(gws.OwsProvider): 

66 def configure(self): 

67 self.alwaysXY = self.cfg('alwaysXY', default=False) 

68 self.forceCrs = gws.gis.crs.get(self.cfg('forceCrs')) 

69 self.maxRequests = self.cfg('maxRequests') 

70 self.operations = [] 

71 self.sourceLayers = [] 

72 self.url = self.cfg('url') 

73 self.version = '' 

74 

75 p = self.cfg('authorization') 

76 self.authorization = gws.OwsAuthorization(p) if p else None 

77 

78 def configure_operations(self, operations_from_caps): 

79 d = {} 

80 

81 for op in operations_from_caps: 

82 d[op.verb] = op 

83 

84 for cfg in (self.cfg('operations') or []): 

85 # add an operation from the config, borrowing missing attributes from a caps op 

86 verb = cfg.get('verb') 

87 caps_op = d.get(verb, {}) 

88 d[verb] = gws.OwsOperation( 

89 verb=verb, 

90 formats=gws.u.first_not_none(cfg.get('formats'), caps_op.get('formats'), []), 

91 params=gws.u.first_not_none(cfg.get('params'), caps_op.get('params'), {}), 

92 postUrl=gws.u.first_not_none(cfg.get('postUrl'), caps_op.get('postUrl'), ''), 

93 url=gws.u.first_not_none(cfg.get('url'), caps_op.get('url'), '') 

94 ) 

95 

96 self.operations = list(d.values()) 

97 

98 for op in self.operations: 

99 op.preferredFormat = self._preferred_format(op) 

100 

101 def _preferred_format(self, op: gws.OwsOperation) -> Optional[str]: 

102 prefer_fmts = _PREFER_IMAGE_MIME if op.verb in _IMAGE_VERBS else _PREFER_XML_MIME 

103 

104 if not op.formats: 

105 return prefer_fmts[0] 

106 

107 # select the "best" (leftmost in the preferred list) format 

108 

109 best_pos = 999 

110 best_fmt = '' 

111 

112 for fmt in op.formats: 

113 canon_fmt = gws.lib.mime.get(fmt) 

114 if not canon_fmt or canon_fmt not in prefer_fmts: 

115 continue 

116 pos = prefer_fmts.index(canon_fmt) 

117 if pos < best_pos: 

118 best_pos = pos 

119 best_fmt = fmt 

120 

121 return best_fmt or op.formats[0] 

122 

123 def get_operation(self, verb, method=None): 

124 for op in self.operations: 

125 if op.verb == verb: 

126 url = op.postUrl if method == gws.RequestMethod.POST else op.url 

127 if url: 

128 return op 

129 

130 def prepare_operation(self, op: gws.OwsOperation, method: gws.RequestMethod = None, params=None) -> request.Args: 

131 args = request.Args( 

132 method=method or gws.RequestMethod.GET, 

133 headers={}, 

134 params={}, 

135 protocol=self.protocol, 

136 verb=op.verb, 

137 version=self.version, 

138 ) 

139 

140 if args.method == gws.RequestMethod.GET: 

141 args.params.update(op.params) 

142 

143 args.url = op.url 

144 if args.method == gws.RequestMethod.POST: 

145 args.url = op.postUrl 

146 

147 allowed = op.allowedParameters or {} 

148 

149 if params: 

150 for name, val in params.items(): 

151 name = name.upper() 

152 if name in allowed and val not in allowed[name]: 

153 raise gws.Error(f'invalid parameter value {val!r} for {name!r}') 

154 args.params[name] = val 

155 

156 if self.authorization and self.authorization.type == 'basic': 

157 b = base64.encodebytes( 

158 gws.u.to_bytes(self.authorization.username) + b':' + gws.u.to_bytes(self.authorization.password)) 

159 args.headers['Authorization'] = 'Basic ' + gws.u.to_str(b).strip() 

160 

161 return args 

162 

163 def get_capabilities(self): 

164 url, params = gws.lib.net.extract_params(self.url) 

165 op = gws.OwsOperation( 

166 formats=[gws.lib.mime.XML], 

167 url=url, 

168 params=params, 

169 verb=gws.OwsVerb.GetCapabilities, 

170 ) 

171 args = self.prepare_operation(op) 

172 return request.get_text(args, max_age=self.cfg('capsCacheMaxAge'))