Coverage for gws-app/gws/base/ows/server/service.py: 67%
199 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 12:46 +0200
1"""OWS Service."""
3from typing import Optional, cast
5import gws
6import gws.base.legend
7import gws.base.web
8import gws.config.util
9import gws.lib.bounds
10import gws.lib.crs
11import gws.lib.datetimex
12import gws.lib.extent
13import gws.lib.image
14import gws.base.metadata
15import gws.lib.mime
16import gws.lib.xmlx
17from . import core, error, request
20class Config(gws.ConfigWithAccess):
21 """Configuration for OWS services."""
23 defaultFeatureCount: int = 1000
24 """Default number of features per page."""
25 extent: Optional[gws.Extent]
26 """Service extent."""
27 extent: Optional[gws.Extent]
28 """Service extent, in extentCrs or the first supported CRS."""
29 extentCrs: Optional[gws.CrsName]
30 """Service extent CRS."""
31 imageFormats: Optional[list[gws.lib.image.FormatConfig]]
32 """Supported image formats."""
33 maxFeatureCount: int = 10000
34 """Max number of features per page."""
35 metadata: Optional[gws.base.metadata.Config]
36 """Service metadata."""
37 rootLayerUid: str = ''
38 """Root layer uid."""
39 searchTolerance: gws.UomValueStr = '10px'
40 """Search pixel tolerance."""
41 supportedCrs: Optional[list[gws.CrsName]]
42 """List of CRS supported by this service."""
43 templates: Optional[list[gws.ext.config.template]]
44 """XML and HTML templates."""
45 updateSequence: Optional[str]
46 """Service update sequence."""
47 withInspireMeta: bool = False
48 """Emit INSPIRE Metadata."""
49 withStrictParams: bool = False
50 """Use strict params checking."""
53class Object(gws.OwsService):
54 """Baseclass for OWS services."""
56 def configure(self):
57 self.project = self.find_closest(gws.ext.object.project)
59 self.updateSequence = self.cfg('updateSequence')
60 self.withInspireMeta = self.cfg('withInspireMeta')
61 self.withStrictParams = self.cfg('withStrictParams')
63 self.maxFeatureCount = self.cfg('maxFeatureCount')
64 self.defaultFeatureCount = self.cfg('defaultFeatureCount')
65 self.searchTolerance = self.cfg('searchTolerance')
67 self.configure_bounds()
68 self.configure_image_formats()
69 self.configure_templates()
70 self.configure_operations()
71 self.configure_metadata()
73 def configure_image_formats(self):
74 p = self.cfg('imageFormats')
75 if p:
76 self.imageFormats = []
77 for cfg in p:
78 self.imageFormats.append(
79 gws.ImageFormat(
80 mimeTypes=[s.replace(' ', '') for s in cfg.get('mimeTypes', [])],
81 options=cfg.get('options') or {},
82 )
83 )
84 return
86 self.imageFormats = [
87 gws.ImageFormat(mimeTypes=[gws.lib.mime.PNG], options={}),
88 gws.ImageFormat(mimeTypes=[gws.lib.mime.JPEG], options={}),
89 ]
91 def configure_bounds(self):
92 p = self.cfg('supportedCrs')
93 if p:
94 crs_list = [gws.lib.crs.require(s) for s in p]
95 elif self.project:
96 crs_list = [self.project.map.bounds.crs]
97 else:
98 crs_list = [gws.lib.crs.WEBMERCATOR, gws.lib.crs.WGS84]
100 p = self.cfg('extent')
101 if p:
102 q = self.cfg('extentCrs')
103 crs = gws.lib.crs.require(q) if q else crs_list[0]
104 bounds = gws.Bounds(crs=crs, extent=gws.lib.extent.from_list(p))
105 elif self.project:
106 bounds = self.project.map.bounds
107 else:
108 bounds = gws.Bounds(crs=crs_list[0], extent=crs_list[0].extent)
110 self.supportedBounds = [gws.lib.bounds.transform(bounds, crs) for crs in crs_list]
111 return True
113 def configure_templates(self):
114 return gws.config.util.configure_templates_for(self)
116 def configure_operations(self):
117 pass
119 def available_formats(self, verb: gws.OwsVerb):
120 fs = set()
122 if verb in core.IMAGE_VERBS:
123 for fmt in self.imageFormats:
124 fs.update(fmt.mimeTypes)
125 else:
126 for tpl in self.templates:
127 if tpl.subject == f'ows.{verb}':
128 fs.update(tpl.mimeTypes)
130 return sorted(fs)
132 def configure_metadata(self):
133 self.metadata = gws.base.metadata.from_args(
134 self.project.metadata if self.project else None,
135 self.root.app.metadata,
136 self.cfg('metadata'),
137 )
138 return True
140 def post_configure(self):
141 self.post_configure_root_layer()
142 self.post_configure_host()
144 def post_configure_host(self):
145 site = self.root.app.webMgr.site
146 if not site.canonicalHost and not site.hostnames:
147 gws.log.warning(f'{self}: neither "web.site.canonicalHost" nor "web.site.hostnames" is set, service urls will reflect the request host')
149 def post_configure_root_layer(self):
150 self.rootLayer = None
152 uid = self.cfg('rootLayerUid')
153 if not uid:
154 return
156 self.rootLayer = cast(gws.Layer, self.root.get(uid, gws.ext.object.layer))
157 if not self.rootLayer:
158 raise gws.ConfigurationError(f'root layer {uid!r} not found')
160 prj = cast(gws.Project, self.rootLayer.find_closest(gws.ext.object.project))
161 if prj and not self.project:
162 self.project = prj
163 return
165 if self.project != prj:
166 raise gws.ConfigurationError(f'root layer {uid!r} does not belong to {self.project!r}')
168 ##
170 def url_path(self, sr: request.Object) -> str:
171 if sr.project and sr.project != self.project:
172 return gws.u.action_url_path('owsService', serviceUid=self.uid, projectUid=sr.project.uid)
173 else:
174 return gws.u.action_url_path('owsService', serviceUid=self.uid)
176 ##
178 def init_request(self, req: gws.WebRequester) -> request.Object:
179 if req.method == 'GET':
180 return request.Object(self, req, req.params())
182 if req.method == 'POST' and gws.lib.mime.get(req.contentType) == gws.lib.mime.XML:
183 try:
184 xml = gws.lib.xmlx.from_string(req.text(), gws.XmlOptions(removeNamespaces=True))
185 except gws.lib.xmlx.Error:
186 raise gws.base.web.error.BadRequest()
188 is_soap = False
189 if xml.name == 'Envelope':
190 is_soap = True
191 try:
192 xml = xml.findfirst('Body').findfirst()
193 except gws.lib.xmlx.Error:
194 raise gws.base.web.error.BadRequest()
196 params = self.parse_xml_request(xml)
197 if not params:
198 raise gws.base.web.error.BadRequest()
199 return request.Object(self, req, params, xml_element=xml, is_soap=is_soap)
201 # @TODO support application/x-www-form-urlencoded
202 raise gws.base.web.error.BadRequest()
204 def parse_xml_request(self, xml: gws.XmlElement) -> Optional[dict]:
205 return {}
207 def handle_request(self, req: gws.WebRequester) -> gws.ContentResponse:
208 try:
209 sr = self.init_request(req)
210 return self.dispatch_request(sr)
211 except Exception as exc:
212 err = error.from_exception(exc)
213 # @TODO INIMAGE Exceptions
214 # verb = req.param('REQUEST')
215 # if verb in core.IMAGE_VERBS:
216 # return err.to_image_response()
217 return err.to_xml_response('ows' if self.isOwsCommon else 'ogc')
219 def dispatch_request(self, sr: request.Object):
220 fn = getattr(self, sr.operation.handlerName)
221 return fn(sr)
223 def get_template(self, sr: request.Object, mime: str = '') -> Optional[gws.Template]:
224 """Find a template for the given service request."""
225 return self.root.app.templateMgr.find_template(
226 f'ows.{sr.operation.verb}',
227 where=[self, sr.project],
228 user=sr.req.user,
229 mime=mime,
230 )
232 def template_response(self, sr: request.Object, mime: str = '', **kwargs) -> gws.ContentResponse:
233 """Render a template for the given service request."""
235 tpl = self.get_template(sr, mime=mime)
236 if not tpl:
237 # OGC 06-042, 7.2.3.1
238 # If the request specifies a format not supported by the server, the server shall respond with the default text/xml format.
239 gws.log.debug(f'no template: {sr.operation.verb=} {mime=}')
240 if mime == gws.lib.mime.XML:
241 raise error.InvalidFormat()
242 return self.template_response(sr, gws.lib.mime.XML, **kwargs)
244 args = request.TemplateArgs(
245 sr=sr,
246 service=self,
247 serviceUrl=sr.req.canonical_url_for(self.url_path(sr)),
248 url_for=sr.req.canonical_url_for,
249 version=sr.version,
250 intVersion=int(sr.version.replace('.', '')),
251 **kwargs,
252 )
254 return tpl.render(gws.TemplateRenderInput(args=args))
256 def xml_response(self, el: gws.XmlElement, opts: gws.XmlOptions = None) -> gws.ContentResponse:
257 xml = el.to_string(opts)
258 return gws.ContentResponse(mime=gws.lib.mime.XML, content=xml)
260 def image_response(self, sr: request.Object, img: Optional[gws.Image], mime: str) -> gws.ContentResponse:
261 ifmt = self.find_image_format(mime)
262 if img:
263 gws.log.debug(f'image_response: {img.mode()=} {img.size()=} {mime=} {ifmt.options}')
264 content = img.to_bytes(mime, ifmt.options) if img else gws.lib.image.empty_pixel(mime)
265 return gws.ContentResponse(mime=mime, content=content)
267 def find_image_format(self, mime: str) -> gws.ImageFormat:
268 if not mime:
269 return self.imageFormats[0]
270 for f in self.imageFormats:
271 if mime in f.mimeTypes:
272 return f
273 raise error.InvalidFormat()
275 def render_legend(self, sr: request.Object, lcs: list[core.LayerCaps], mime: str) -> gws.ContentResponse:
276 uids = [lc.layer.uid for lc in lcs]
277 cache_key = 'gws.base.ows.server.legend.' + gws.u.sha256(uids) + mime
279 def _get():
280 legend = cast(
281 gws.Legend,
282 self.root.create_temporary(
283 gws.ext.object.legend,
284 type='combined',
285 layerUids=uids,
286 ),
287 )
288 lro = legend.render()
289 if not lro:
290 return self.image_response(sr, None, mime)
291 return self.image_response(sr, gws.base.legend.output_to_image(lro), mime)
293 return gws.u.get_app_global(cache_key, _get)
295 def feature_collection(self, sr: request.Object, lcs: list[core.LayerCaps], hits: int, results: list[gws.SearchResult]) -> core.FeatureCollection:
296 fc = core.FeatureCollection(
297 members=[],
298 numMatched=hits,
299 numReturned=len(results),
300 timestamp=gws.lib.datetimex.to_iso_string(with_tz=':'),
301 )
303 lcs_map = {id(lc.layer): lc for lc in lcs}
305 for r in results:
306 r.feature.transform_to(sr.targetCrs)
307 fc.members.append(
308 core.FeatureCollectionMember(
309 feature=r.feature,
310 layer=r.layer,
311 layerCaps=lcs_map.get(id(r.layer)) if r.layer else None,
312 )
313 )
315 return fc