Coverage for gws-app/gws/plugin/ows_server/csw/__init_

1"""CSW service.

3@TODO to be implemented

5"""

7# from typing import Optional, cast

9# import gws

10# import gws.base.web

11# import gws.lib.datetimex

12# import gws.gis.extent

13# import gws.lib.metadata

14# import gws.lib.mime

15# import gws.lib.xmlx as xmlx

16# import gws.base.ows.server

17#

18# from . import filter

19#

20# gws.ext.new.owsService('csw')

21#

22#

23# class Profile(gws.Enum):

24# ISO = 'ISO'

25# DCMI = 'DCMI'

26#

27#

28# class Config(gws.base.ows.server.service.Config):

29# """CSW Service configuration"""

30# # @TODO no support for DCMI yet

31# # profile: Profile = Profile.ISO

32# """metadata profile"""

33# pass

34#

35#

36# class Object(gws.base.ows.server.service.Object):

37# protocol = gws.OwsProtocol.CSW

38# supportedVersions = ['2.0.2']

39#

40# records: dict[str, gws.Metadata]

41# index: list

42# profile: Profile

43#

44# @property

45# def default_templates(self):

46# base = gws.u.dirname(__file__) + '/templates'

47# return [

48# gws.Config(

49# type='py',

50# path=f'{base}/getCapabilities.cx.py',

51# subject='ows.GetCapabilities',

52# mimeTypes=['xml'],

53# access=gws.c.PUBLIC,

54# ),

55# gws.Config(

56# type='py',

57# path=f'{base}/getRecords.cx.py',

58# subject='ows.getRecords',

59# mimeTypes=['xml'],

60# access=gws.c.PUBLIC,

61# ),

62# gws.Config(

63# type='py',

64# path=f'{base}/getRecordById.cx.py',

65# subject='ows.GetRecordById',

66# mimeTypes=['xml'],

67# access=gws.c.PUBLIC,

68# ),

69# gws.Config(

70# type='py',

71# path=f'{base}/record.cx.py',

72# subject='ows.Record',

73# mimeTypes=['xml'],

74# access=gws.c.PUBLIC,

75# ),

76# ]

77#

78# @property

79# def default_metadata(self):

80# return gws.Data(

81# inspireDegreeOfConformity='notEvaluated',

82# inspireMandatoryKeyword='humanCatalogueViewer',

83# inspireResourceType='service',

84# inspireSpatialDataServiceType='discovery',

85# isoScope='dataset',

86# isoSpatialRepresentationType='vector',

87# )

88#

89# ##

90#

91# def configure(self):

92# self.records = {}

93# self.index = []

94# self.profile = Profile.ISO

95#

96# def post_configure(self):

97# self._collect_metadata()

98# self._create_index()

99# cnt = len(self.records)

100# gws.log.info(f'CSW service configured with {cnt} records')

101#

102# ##

103#

104# def handle_request(self, req: gws.WebRequester) -> gws.ContentResponse:

105# rd = core.Request(req=req, project=None, service=self)

106#

107# if req.method == 'GET':

108# return self.dispatch_request(rd, req.param('request', default='record'))

109#

110# # CSW should accept POST'ed xml, which can be wrapped in a SOAP envelope

111#

112# try:

113# rd.xml_element = xmlx.from_string(req.text)

114# except xmlx.Error:

115# raise gws.base.web.error.BadRequest()

116#

117# if rd.xml_element.name.lower() == 'envelope':

118# rd.xml_is_soap = True

119# try:

120# rd.xml_element = xmlx.first(xmlx.first('body'))

121# except Exception:

122# raise gws.base.web.error.BadRequest()

123#

124# return self.dispatch_request(rd, xmlx.unqualify_name(rd.xml_element.name.lower()))

125#

126# def handle_getcapabilities(self, rd: core.Request):

127# return self.template_response(rd, gws.OwsVerb.GetCapabilities, context={

128# 'profile': self.profile,

129# 'version': self.request_version(rd),

130# })

131#

132# def handle_describerecord(self, rd: core.Request):

133# xml = gws.u.read_file(gws.u.dirname(__file__) + '/templates/describeRecord.xml')

134# return gws.ContentResponse(mime=gws.lib.mime.XML, content=xml)

135#

136# def handle_getrecords(self, rd: core.Request):

137# records = self._find_records(rd)

138#

139# results = {

140# 'timestamp': gws.lib.datetimex.now(),

141# 'next': 0,

142# 'count_total': len(records),

143# 'count_return': len(records),

144# }

145#

146# return self.template_response(rd, gws.OwsVerb.GetRecords, context={

147# 'records': [md.values for md in records],

148# 'results': results,

149# 'with_soap': rd.xml_is_soap,

150# 'profile': self.profile,

151# 'version': self.request_version(rd),

152# })

153#

154# def handle_getrecordbyid(self, rd: core.Request):

155# md = self.records.get(rd.req.param('id'))

156# if not md:

157# raise gws.base.web.error.NotFound()

158#

159# return self.template_response(rd, gws.OwsVerb.GetRecordById, context={

160# 'record': md.values,

161# 'with_soap': rd.xml_is_soap,

162# 'profile': self.profile,

163# 'version': self.request_version(rd),

164# })

165#

166# def handle_record(self, rd: core.Request):

167# # Record is our internal method to return bare metadata without the GetRecordByIdResponse envelope

168# # see _make_link below

169#

170# md = self.records.get(rd.req.param('id'))

171# if not md:

172# raise gws.base.web.error.NotFound()

173#

174# return self.template_response(rd, cast(gws.OwsVerb, 'Record'), context={

175# 'record': md.values,

176# 'with_soap': False,

177# 'profile': self.profile,

178# 'version': self.request_version(rd),

179# })

180#

181# ##

182#

183# def _collect_metadata(self):

184# # collect objects whose metadata should be published in the catalog

185# #

186# # - object should have `metadata`

187# # - object must be public

188# # - `metadata` should have `catalogUid`

189# # - `metadata.metaLinks` should be empty

190# #

191# # `metadata.metaLinks[0]` will be set to our csw url

192#

193# self.records = {}

194#

195# for obj in self.root.find_all():

196# md: gws.lib.metadata.Metadata = gws.u.get(obj, 'metadata')

197#

198# if not md or not md.get('catalogUid'):

199# continue

200#

201# cid = gws.u.to_uid(md.get('catalogUid'))

202#

203# if md.get('metaLinks'):

204# gws.log.debug(f'csw: skip {cid}: has metalinks')

205# continue

206#

207# if not gws.u.is_public_object(obj):

208# gws.log.debug(f'csw: skip {cid}: not public')

209# continue

210#

211# md.set('catalogUid', cid)

212# md.set('catalogCitationUid', cid)

213# md.set('metaLinks', [self._make_link(cid)])

214#

215# extent = gws.u.get(obj, 'extent') or gws.u.get(obj, 'map.extent')

216# crs = gws.u.get(obj, 'crs') or gws.u.get(obj, 'map.crs')

217# if extent and crs:

218# md.set('wgsExtent', gws.gis.extent.transform_to_4326(extent, crs))

219# md.set('crs', crs)

220# # @TODO get boundingPolygonElement somehow

221#

222# self.records[cid] = md

223#

224# def _make_link(self, cid):

225# return gws.MetadataLink(

226# url=gws.u.action_url_path('owsService', serviceUid=self.uid, request='record', id=cid),

227# format=gws.lib.mime.XML,

228# type='TC211' if self.profile == 'ISO' else 'DCMI'

229# )

230#

231# def _create_index(self):

232# self.index = []

233#

234# for uid, md in self.records.items():

235# s = gws.u.get(md, 'title')

236# if s:

237# self.index.append(['title', s, s.lower(), uid])

238# s = gws.u.get(md, 'abstract')

239# if s:

240# self.index.append(['abstract', s, s.lower(), uid])

241# s = gws.u.get(md, 'keywords')

242# if s:

243# for kw in s:

244# self.index.append(('subject', kw, kw.lower(), uid))

245#

246# def _find_records(self, rd: core.Request):

247# flt = None

248# if rd.xml_element:

249# flt = xmlx.first(xmlx.first('Query.Constraint.Filter'))

250# if not flt:

251# return self.records.values()

252# f = filter.Filter(self.index)

253# return f.apply(flt, self.records.values())

Coverage for gws-app/gws/plugin/ows_server/csw/init.py: 100%

0 statements