Coverage for pygeoapi/api/stac.py: 11%

1# =================================================================

3# Authors: Tom Kralidis <tomkralidis@gmail.com>

4# Francesco Bartoli <xbartolone@gmail.com>

5# Sander Schaminee <sander.schaminee@geocat.net>

6# John A Stevenson <jostev@bgs.ac.uk>

7# Colin Blackburn <colb@bgs.ac.uk>

8# Ricardo Garcia Silva <ricardo.garcia.silva@geobeyond.it>

9# Bernhard Mallinger <bernhard.mallinger@eox.at>

10#

16#

17# Permission is hereby granted, free of charge, to any person

18# obtaining a copy of this software and associated documentation

19# files (the "Software"), to deal in the Software without

20# restriction, including without limitation the rights to use,

21# copy, modify, merge, publish, distribute, sublicense, and/or sell

22# copies of the Software, and to permit persons to whom the

23# Software is furnished to do so, subject to the following

24# conditions:

25#

26# The above copyright notice and this permission notice shall be

27# included in all copies or substantial portions of the Software.

28#

29# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,

30# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES

31# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND

32# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT

33# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,

34# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING

35# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR

36# OTHER DEALINGS IN THE SOFTWARE.

37#

38# =================================================================

40from copy import deepcopy

41from http import HTTPStatus

42import json

43import logging

44from typing import Any, Tuple, Union

45from urllib.parse import urlencode

47from shapely import from_geojson

49from pygeoapi import l10n

50from pygeoapi.formats import FORMAT_TYPES, F_JSON, F_HTML

51from pygeoapi import api as ogc_api

52from pygeoapi.api import itemtypes as itemtypes_api

53from pygeoapi.plugin import load_plugin

55from pygeoapi.provider import get_provider_by_type

56from pygeoapi.provider.base import (

57 ProviderConnectionError, ProviderNotFoundError, ProviderTypeError

58)

59from pygeoapi.util import (filter_dict_by_key_value, get_current_datetime,

60 render_j2_template, to_json)

62from . import APIRequest, API

65LOGGER = logging.getLogger(__name__)

68CONFORMANCE_CLASSES = []

71# TODO: no tests for this?

72def get_stac_root(api: API, request: APIRequest) -> Tuple[dict, int, str]:

73 """

74 Provide STAC root page

76 :param request: APIRequest instance with query params

78 :returns: tuple of headers, status code, content

79 """

80 headers = request.get_response_headers(**api.api_headers)

82 id_ = 'pygeoapi-stac'

83 stac_version = '1.0.0-rc.2'

84 stac_url = f'{api.base_url}/stac'

86 content = {

87 'id': id_,

88 'type': 'Catalog',

89 'stac_version': stac_version,

90 'title': l10n.translate(

91 api.config['metadata']['identification']['title'],

92 request.locale),

93 'description': l10n.translate(

94 api.config['metadata']['identification']['description'],

95 request.locale),

96 'links': []

97 }

99 stac_collections = filter_dict_by_key_value(api.config['resources'],

100 'type', 'stac-collection')

101

102 for key, value in stac_collections.items():

103 try:

104 _ = load_plugin('provider', get_provider_by_type(

105 value['providers'], 'stac'))

106 except ProviderTypeError:

107 LOGGER.debug('Not a STAC-based provider; skipping')

108 continue

109

110 content['links'].append({

111 'rel': 'child',

112 'href': f'{stac_url}/{key}?f={F_JSON}',

113 'type': FORMAT_TYPES[F_JSON],

114 'title': key,

115 'description': value['description']

116 })

117 content['links'].append({

118 'rel': 'child',

119 'href': f'{stac_url}/{key}',

120 'type': FORMAT_TYPES[F_HTML],

121 'title': key,

122 'description': value['description']

123 })

124

125 if request.format == F_HTML: # render

126 content = render_j2_template(

127 api.tpl_config, api.config['server']['templates'],

128 'stac/collection.html', content, request.locale)

129

130 return headers, HTTPStatus.OK, content

131

132 return headers, HTTPStatus.OK, to_json(content, api.pretty_print)

133

134

135# TODO: no tests for this?

136def get_stac_path(api: API, request: APIRequest,

137 path: str) -> Tuple[dict, int, str]:

138 """

139 Provide STAC resource path

140

141 :param request: APIRequest instance with query params

142

143 :returns: tuple of headers, status code, content

144 """

145 headers = request.get_response_headers(**api.api_headers)

146

147 dataset = None

148 LOGGER.debug(f'Path: {path}')

149 dir_tokens = path.split('/')

150 if dir_tokens:

151 dataset = dir_tokens[0]

152

153 stac_collections = filter_dict_by_key_value(api.config['resources'],

154 'type', 'stac-collection')

155

156 if dataset not in stac_collections:

157 msg = 'Collection not found'

158 return api.get_exception(HTTPStatus.NOT_FOUND, headers,

159 request.format, 'NotFound', msg)

160

161 LOGGER.debug('Loading provider')

162 try:

163 p = load_plugin('provider', get_provider_by_type(

164 stac_collections[dataset]['providers'], 'stac'))

165 except ProviderConnectionError:

166 msg = 'connection error (check logs)'

167 return api.get_exception(

168 HTTPStatus.INTERNAL_SERVER_ERROR, headers,

169 request.format, 'NoApplicableCode', msg)

170

171 id_ = f'{dataset}-stac'

172 stac_version = '1.0.0-rc.2'

173

174 content = {

175 'id': id_,

176 'type': 'Catalog',

177 'stac_version': stac_version,

178 'description': l10n.translate(

179 stac_collections[dataset]['description'], request.locale),

180 'links': []

181 }

182 try:

183 stac_data = p.get_data_path(

184 f'{api.base_url}/stac',

185 path,

186 path.replace(dataset, '', 1)

187 )

188 except ProviderNotFoundError:

189 msg = 'resource not found'

190 return api.get_exception(HTTPStatus.NOT_FOUND, headers,

191 request.format, 'NotFound', msg)

192 except Exception:

193 msg = 'data query error'

194 return api.get_exception(

195 HTTPStatus.INTERNAL_SERVER_ERROR, headers,

196 request.format, 'NoApplicableCode', msg)

197

198 if isinstance(stac_data, dict):

199 content.update(stac_data)

200 content['links'].extend(

201 stac_collections[dataset].get('links', []))

202

203 if request.format == F_HTML: # render

204 content['path'] = path

205 if 'assets' in content: # item view

206 if content['type'] == 'Collection':

207 content = render_j2_template(

208 api.tpl_config, api.config['server']['templates'],

209 'stac/collection_base.html',

210 content, request.locale)

211 elif content['type'] == 'Feature':

212 content = render_j2_template(

213 api.tpl_config, api.config['server']['templates'],

214 'stac/item.html', content, request.locale)

215 else:

216 msg = f'Unknown STAC type {content.type}'

217 return api.get_exception(

218 HTTPStatus.INTERNAL_SERVER_ERROR,

219 headers,

220 request.format,

221 'NoApplicableCode',

222 msg)

223 else:

224 content = render_j2_template(

225 api.tpl_config, api.config['server']['templates'],

226 'stac/catalog.html', content, request.locale)

227

228 return headers, HTTPStatus.OK, content

229

230 return headers, HTTPStatus.OK, to_json(content, api.pretty_print)

231

232 else: # send back file

233 headers.pop('Content-Type', None)

234 return headers, HTTPStatus.OK, stac_data

235

236

237def landing_page(api: API,

238 request: APIRequest) -> Tuple[dict, int, str]:

239 """

240 Provide API landing page

241

242 :param request: A request object

243

244 :returns: tuple of headers, status code, content

245 """

246

247 request._format = F_JSON

248

249 headers, status, content = ogc_api.landing_page(api, request)

250

251 content = json.loads(content)

252

253 content['id'] = 'pygeoapi-catalogue'

254 content['stac_version'] = '1.0.0'

255 content['conformsTo'] = [

256 'https://api.stacspec.org/v1.0.0/core',

257 'https://api.stacspec.org/v1.0.0/item-search',

258 'https://api.stacspec.org/v1.0.0/item-search#sort'

259 ]

260 content['type'] = 'Catalog'

261

262 content['links'] = [{

263 'rel': request.get_linkrel(F_JSON),

264 'type': FORMAT_TYPES[F_JSON],

265 'title': l10n.translate('This document as JSON', request.locale),

266 'href': f"{api.base_url}/stac-api?f={F_JSON}"

267 }, {

268 'rel': 'root',

269 'type': FORMAT_TYPES[F_JSON],

270 'title': l10n.translate('This document as JSON', request.locale),

271 'href': f"{api.base_url}/stac-api?f={F_JSON}"

272 }, {

273 'rel': 'service-desc',

274 'type': 'application/vnd.oai.openapi+json;version=3.0',

275 'title': l10n.translate('The OpenAPI definition as JSON', request.locale), # noqa

276 'href': f"{api.base_url}/openapi"

277 }, {

278 'rel': 'service-doc',

279 'type': FORMAT_TYPES[F_HTML],

280 'title': l10n.translate('The OpenAPI definition as HTML', request.locale), # noqa

281 'href': f"{api.base_url}/openapi?f={F_HTML}",

282 'hreflang': api.default_locale

283 }, {

284 'rel': 'search',

285 'type': FORMAT_TYPES[F_JSON],

286 'title': l10n.translate('STAC API search', request.locale),

287 'href': f"{api.base_url}/stac-api//search?f={F_JSON}"

288 }]

289

290 return headers, status, to_json(content, api.pretty_print)

291

292

293def search(api: API, request: Union[APIRequest, Any]) -> Tuple[dict, int, str]:

294 """

295 STAC API Queries stac-collection

296

297 :param request: A request object

298 :param dataset: dataset name

299

300 :returns: tuple of headers, status code, content

301 """

302

303 stac_api_collections = {}

304

305 request._format = F_JSON

306

307 headers = request.get_response_headers(**api.api_headers)

308

309 LOGGER.debug('Checking for STAC collections')

310 collections = filter_dict_by_key_value(api.config['resources'],

311 'type', 'stac-collection')

312

313 if not collections:

314 return api.get_exception(

315 HTTPStatus.NOT_IMPLEMENTED, headers, F_JSON, 'NotImplemented',

316 'No configured STAC searchable collection')

317

318 LOGGER.debug('Checking for STAC collections with features or records')

319 for key, value in collections.items():

320 found_collection = False

321 for fr in ['feature', 'record']:

322 try:

323 _ = get_provider_by_type(value['providers'], fr)

324 found_collection = True

325 break

326 except ProviderTypeError:

327 pass

328

329 if found_collection:

330 stac_api_collections[key] = value

331

332 if not stac_api_collections:

333 msg = 'No STAC API collections configured'

334 return api.get_exception(HTTPStatus.INTERNAL_SERVER_ERROR, headers,

335 request.format, 'NotApplicable', msg)

336

337 if request.data:

338 LOGGER.debug('Intercepting STAC POST request into query args')

339 request_data = json.loads(request.data)

340 request_params = deepcopy(dict(request.params))

341

342 for qp in ['bbox', 'datetime', 'limit', 'offset']:

343 if qp in request_data:

344 if qp == 'bbox' and isinstance(request_data[qp], list):

345 request_params[qp] = ','.join(str(b) for b in request_data[qp]) # noqa

346 else:

347 request_params[qp] = request_data[qp]

348

349 request._args = request_params

350 request._data = None

351

352 stac_api_response = {

353 'type': 'FeatureCollection',

354 'features': [],

355 'numberMatched': 0,

356 'links': []

357 }

358

359 for key, value in stac_api_collections.items():

360 api.config['resources'][key]['type'] = 'collection'

361 headers, status, content = itemtypes_api.get_collection_items(

362 api, request, key)

363 api.config['resources'][key]['type'] = 'stac-collection'

364

365 if status != HTTPStatus.OK:

366 return headers, status, to_json(content, api.pretty_print)

367

368 content = json.loads(content)

369 stac_api_response['numberMatched'] += content.get('numberMatched', 0)

370

371 if len(content.get('features', [])) > 0:

372 for feature in content['features']:

373 if 'stac_version' not in feature:

374 feature['stac_version'] = '1.0.0'

375 feature['properties'].update(get_temporal(feature))

376 stac_api_response['features'].append(feature)

377

378 if feature.get('geometry') is not None and 'bbox' not in feature: # noqa

379 geom = from_geojson(json.dumps(feature['geometry']))

380 feature['bbox'] = geom.bounds

381

382 for la in ['links', 'assets']:

383 if feature.get(la) is None:

384 feature[la] = []

385

386 stac_api_response['numberReturned'] = len(stac_api_response['features'])

387

388 stac_api_response['links'].append({

389 'rel': 'root',

390 'type': FORMAT_TYPES[F_JSON],

391 'title': l10n.translate('STAC API landing page', request.locale),

392 'href': f"{api.base_url}/stac-api?f={F_JSON}"

393 })

394

395 LOGGER.debug('Generating paging links')

396

397 next_link = False

398 prev_link = False

399 request_params = deepcopy(dict(request._args))

400 limit = itemtypes_api.evaluate_limit(

401 request_params.get('limit'),

402 api.config['server'].get('limits', {}), {})

403 offset = int(request_params.get('offset', 0))

404

405 if stac_api_response.get('numberMatched', -1) > (limit + offset):

406 next_link = True

407 elif len(stac_api_response['features']) == limit:

408 next_link = True

409

410 if offset > 0:

411 prev_link = True

412

413 if prev_link:

414 request_params['offset'] = max(0, offset - limit)

415 if request_params['offset'] == 0:

416 request_params.pop('offset')

417

418 request_params_qs = urlencode(request_params)

419

420 stac_api_response['links'].append({

421 'rel': 'prev',

422 'type': FORMAT_TYPES[F_JSON],

423 'title': l10n.translate('Items (prev)', request.locale),

424 'href': f"{api.base_url}/stac-api/search?{request_params_qs}"

425 })

426

427 if next_link:

428 request_params['offset'] = offset + limit

429 request_params_qs = urlencode(request_params)

430

431 stac_api_response['links'].append({

432 'rel': 'next',

433 'type': FORMAT_TYPES[F_JSON],

434 'title': l10n.translate('Items (next)', request.locale),

435 'href': f"{api.base_url}/stac-api/search?{request_params_qs}"

436 })

437

438 return headers, HTTPStatus.OK, to_json(stac_api_response, api.pretty_print)

439

440

441def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str, dict]]: # noqa

442 """

443 Get OpenAPI fragments

444

445 :param cfg: `dict` of configuration

446 :param locale: `str` of locale

447

448 :returns: `tuple` of `list` of tag objects, and `dict` of path objects

449 """

450

451 LOGGER.debug('setting up STAC')

452 stac_collections = filter_dict_by_key_value(cfg['resources'],

453 'type', 'stac-collection')

454 paths = {}

455 if stac_collections: 455 ↛ 456line 455 didn't jump to line 456 because the condition on line 455 was never true

456 paths['/stac/catalog'] = {

457 'get': {

458 'summary': 'SpatioTemporal Asset Catalog',

459 'description': 'SpatioTemporal Asset Catalog',

460 'tags': ['stac'],

461 'operationId': 'getStacCatalog',

462 'parameters': [],

463 'responses': {

464 '200': {'$ref': '#/components/responses/200'},

465 'default': {'$ref': '#/components/responses/default'}

466 }

467 }

468 }

469 return [{'name': 'stac'}], {'paths': paths}

470

471

472def get_temporal(feature: dict) -> dict:

473 """

474 Helper function to try and derive a useful temporal

475 definition on a non-STAC item

476

477 :param feature: `dict` of GeoJSON feature

478

479 :returns: `dict` of `datetime` or `start_datetime` and `end_datetime`

480 """

481

482 value = {}

483

484 datetime_ = feature['properties'].get('datetime')

485 start_datetime = feature['properties'].get('start_datetime')

486 end_datetime = feature['properties'].get('end_datetime')

487

488 if datetime_ is None and None not in [start_datetime, end_datetime]:

489 LOGGER.debug('Temporal range partially exists')

490 elif datetime_ is not None:

491 LOGGER.debug('Temporal instant exists')

492

493 LOGGER.debug('Attempting to derive temporal from GeoJSON feature')

494 LOGGER.debug(feature)

495 if feature.get('time') is not None:

496 if feature['time'].get('timestamp') is not None:

497 value['datetime'] = feature['time']['timestamp']

498 if feature['time'].get('interval') is not None:

499 value['start_datetime'] = feature['time']['interval'][0]

500 value['end_datetime'] = feature['time']['interval'][1]

501

502 if feature['properties'].get('created') is not None:

503 value['datetime'] = feature['properties']['created']

504

505 if not value:

506 value['datetime'] = get_current_datetime()

507

508 return value