Coverage for gws-app/gws/base/ows/server/action.py: 53%

1"""OWS server action.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.base.action 

7import gws.base.web 

8import gws.lib.mime 

9import gws.lib.xmlx 

10 

11from . import core, layer_caps, error 

12 

13 

14class GetServiceRequest(gws.Request): 

15 """Request to an OWS service.""" 

16 

17 serviceUid: str 

18 """Service uid.""" 

19 

20 

21class GetSchemaRequest(gws.Request): 

22 """Request for an XML schema.""" 

23 

24 namespace: str 

25 """Namespace prefix, optionally with the ``.xsd`` extension.""" 

26 

27 

28@gws.ext.config.action('ows') 

29class Config(gws.base.action.Config): 

30 """Action that serves the configured OWS services.""" 

31 

32 

33@gws.ext.object.action('ows') 

34class Object(gws.base.action.Object): 

35 """OWS action. 

36 

37 Dispatches requests to the configured OWS services and serves XML schemas 

38 generated for layer namespaces. 

39 """ 

40 

41 @gws.ext.command.get('owsService') 

42 def get_service(self, req: gws.WebRequester, p: GetServiceRequest) -> gws.ContentResponse: 

43 """Handle an OWS service request.""" 

44 

45 return self._handle_service(req, p) 

46 

47 @gws.ext.command.post('owsService') 

48 def post_service(self, req: gws.WebRequester, p: GetServiceRequest) -> gws.ContentResponse: 

49 """Handle an OWS service request.""" 

50 

51 return self._handle_service(req, p) 

52 

53 def _handle_service(self, req: gws.WebRequester, p: GetServiceRequest) -> gws.ContentResponse: 

54 """Locate the service, check access and pass the request to it.""" 

55 srv = cast(gws.OwsService, self.root.get(p.serviceUid, gws.ext.object.owsService)) 

56 if not srv: 

57 raise gws.NotFoundError(f'{p.serviceUid=} not found') 

58 if not req.user.can_use(srv): 

59 raise gws.ForbiddenError(f'{p.serviceUid=} forbidden') 

60 return srv.handle_request(req) 

61 

62 @gws.ext.command.get('owsXml') 

63 def get_schema(self, req: gws.WebRequester, p: GetSchemaRequest) -> gws.ContentResponse: 

64 """Return an XML schema for the layers with the given namespace prefix.""" 

65 

66 try: 

67 content = self._make_schema(req, p) 

68 except Exception as exc: 

69 return error.from_exception(exc).to_xml_response() 

70 return gws.ContentResponse(mimeType=gws.lib.mime.XML, content=content) 

71 

72 def _make_schema(self, req, p) -> str: 

73 """Create the XML schema document for a namespace.""" 

74 s = p.namespace 

75 if s.endswith('.xsd'): 

76 s = s[:-4] 

77 lcs = [] 

78 

79 for la in self.root.find_all(gws.ext.object.layer): 

80 layer = cast(gws.Layer, la) 

81 if req.user.can_read(layer) and layer.ows.xmlNamespace and layer.ows.xmlNamespace.prefix == s: 

82 lcs.append(layer_caps.for_layer(layer, req.user)) 

83 

84 if not lcs: 

85 raise gws.NotFoundError(f'namespace not found: {p.namespace=}') 

86 

87 el, opts = layer_caps.xml_schema(lcs, req.user) 

88 if not el: 

89 raise gws.NotFoundError(f'cannot create schema: {p.namespace=}') 

90 

91 opts.withNamespaceDeclarations = True 

92 opts.withSchemaLocations = True 

93 opts.withXmlDeclaration = True 

94 

95 return el.to_string(opts)