Coverage for pygeoapi/api/itemtypes.py: 71%

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

2# 

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# 

10# Copyright (c) 2026 Tom Kralidis 

11# Copyright (c) 2025 Francesco Bartoli 

12# Copyright (c) 2022 John A Stevenson and Colin Blackburn 

13# Copyright (c) 2023 Ricardo Garcia Silva 

14# 

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

16# obtaining a copy of this software and associated documentation 

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

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

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

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

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

22# conditions: 

23# 

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

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

26# 

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

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

29# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 

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

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

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

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

34# OTHER DEALINGS IN THE SOFTWARE. 

35# 

36# ================================================================= 

37 

38from collections import ChainMap 

39from copy import deepcopy 

40from datetime import datetime, UTC 

41from http import HTTPStatus 

42import logging 

43from typing import Any, Tuple, Union 

44import urllib.parse 

45 

46from pygeofilter.parsers.ecql import parse as parse_ecql_text 

47from pygeofilter.parsers.cql2_json import parse as parse_cql2_json 

48from pyproj.exceptions import CRSError 

49 

50from pygeoapi import l10n 

51from pygeoapi.api import evaluate_limit 

52from pygeoapi.api.pubsub import publish_message 

53from pygeoapi.crs import (DEFAULT_CRS, DEFAULT_STORAGE_CRS, 

54 create_crs_transform_spec, get_supported_crs_list, 

55 modify_pygeofilter, transform_bbox, 

56 set_content_crs_header) 

57from pygeoapi.formats import F_JSON, FORMAT_TYPES, F_HTML, F_JSONLD 

58from pygeoapi.formatter.base import FormatterSerializationError 

59from pygeoapi.linked_data import geojson2jsonld 

60from pygeoapi.openapi import get_oas_30_parameters 

61from pygeoapi.plugin import load_plugin, PLUGINS 

62from pygeoapi.provider import filter_providers_by_type, get_provider_by_type 

63from pygeoapi.provider.base import ( 

64 ProviderGenericError, ProviderItemNotFoundError, 

65 ProviderTypeError, SchemaType) 

66 

67from pygeoapi.util import (to_json, filter_dict_by_key_value, str2bool, 

68 render_j2_template, get_dataset_formatters) 

69 

70from . import APIRequest, API, SYSTEM_LOCALE, validate_bbox, validate_datetime 

71 

72LOGGER = logging.getLogger(__name__) 

73 

74OGC_RELTYPES_BASE = 'http://www.opengis.net/def/rel/ogc/1.0' 

75 

76 

77CONFORMANCE_CLASSES_FEATURES = [ 

78 'http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/core', 

79 'http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/oas30', 

80 'http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/html', 

81 'http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/geojson', 

82 'http://www.opengis.net/spec/ogcapi-features-2/1.0/conf/crs', 

83 'http://www.opengis.net/spec/ogcapi-features-3/1.0/conf/queryables', 

84 'http://www.opengis.net/spec/ogcapi-features-3/1.0/conf/queryables-query-parameters', # noqa 

85 'http://www.opengis.net/spec/ogcapi-features-4/1.0/conf/create-replace-delete', # noqa 

86 'http://www.opengis.net/spec/ogcapi-features-5/1.0/conf/schemas', 

87 'http://www.opengis.net/spec/ogcapi-features-5/1.0/conf/core-roles-features' # noqa 

88] 

89 

90CONFORMANCE_CLASSES_RECORDS = [ 

91 'http://www.opengis.net/spec/ogcapi-records-1/1.0/conf/core', 

92 'http://www.opengis.net/spec/ogcapi-records-1/1.0/conf/sorting', 

93 'http://www.opengis.net/spec/ogcapi-records-1/1.0/conf/opensearch', 

94 'http://www.opengis.net/spec/ogcapi-records-1/1.0/conf/json', 

95 'http://www.opengis.net/spec/ogcapi-records-1/1.0/conf/html' 

96] 

97 

98 

99def get_collection_queryables(api: API, request: Union[APIRequest, Any], 

100 dataset: str | None = None 

101 ) -> Tuple[dict, int, str]: 

102 """ 

103 Provide collection queryables 

104 

105 :param request: A request object 

106 :param dataset: name of collection 

107 

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

109 """ 

110 

111 domains = {} 

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

113 

114 if any([dataset is None, 114 ↛ 117line 114 didn't jump to line 117 because the condition on line 114 was never true

115 dataset not in api.config['resources'].keys()]): 

116 

117 msg = 'Collection not found' 

118 return api.get_exception( 

119 HTTPStatus.NOT_FOUND, headers, request.format, 'NotFound', msg) 

120 

121 LOGGER.debug('Creating collection queryables') 

122 

123 p = None 

124 for pt in ['feature', 'coverage', 'record']: 124 ↛ 133line 124 didn't jump to line 133 because the loop on line 124 didn't complete

125 try: 

126 LOGGER.debug(f'Loading {pt} provider') 

127 p = load_plugin('provider', get_provider_by_type( 

128 api.config['resources'][dataset]['providers'], pt)) 

129 break 

130 except ProviderTypeError: 

131 LOGGER.debug(f'Providing type {pt} not found') 

132 

133 if p is None: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true

134 msg = 'queryables not available for this collection' 

135 return api.get_exception( 

136 

137 HTTPStatus.BAD_REQUEST, headers, request.format, 

138 'NoApplicableError', msg) 

139 

140 LOGGER.debug('Processing profile') 

141 profile = request.params.get('profile', '') 

142 

143 LOGGER.debug('Processing properties') 

144 val = request.params.get('properties') 

145 if val is not None: 

146 properties = [x for x in val.split(',') if x] 

147 properties_to_check = set(p.properties) | set(p.fields.keys()) 

148 

149 if len(list(set(properties) - set(properties_to_check))) > 0: 

150 msg = 'unknown properties specified' 

151 return api.get_exception( 

152 HTTPStatus.BAD_REQUEST, headers, request.format, 

153 'InvalidParameterValue', msg) 

154 else: 

155 properties = [] 

156 

157 queryables_id = f'{api.get_collections_url()}/{dataset}/queryables' 

158 

159 if request.params: 

160 queryables_id += '?' + urllib.parse.urlencode(request.params) 

161 

162 queryables = { 

163 'type': 'object', 

164 'title': l10n.translate( 

165 api.config['resources'][dataset]['title'], request.locale), 

166 'properties': {}, 

167 '$schema': 'http://json-schema.org/draft/2019-09/schema', 

168 '$id': queryables_id 

169 } 

170 

171 if p.fields: 171 ↛ 177line 171 didn't jump to line 177 because the condition on line 171 was always true

172 queryables['properties']['geometry'] = { 

173 'format': 'geometry-any', 

174 'x-ogc-role': 'primary-geometry' 

175 } 

176 

177 if profile == 'actual-domain': 

178 try: 

179 domains, _ = p.get_domains(properties) 

180 except NotImplementedError: 

181 LOGGER.debug('Domains are not suported by this provider') 

182 domains = {} 

183 

184 for k, v in p.fields.items(): 

185 show_field = False 

186 if properties and k not in properties: 

187 continue 

188 if p.properties: 188 ↛ 189line 188 didn't jump to line 189 because the condition on line 188 was never true

189 if k in p.properties: 

190 show_field = True 

191 else: 

192 show_field = True 

193 

194 if show_field: 194 ↛ 184line 194 didn't jump to line 184 because the condition on line 194 was always true

195 queryables['properties'][k] = { 

196 'title': k, 

197 'type': v['type'] 

198 } 

199 if v['type'] == 'float': 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true

200 queryables['properties'][k]['type'] = 'number' 

201 if v.get('format') is not None: 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true

202 queryables['properties'][k]['format'] = v['format'] 

203 if 'values' in v: 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true

204 queryables['properties'][k]['enum'] = v['values'] 

205 

206 if k == p.id_field: 

207 queryables['properties'][k]['x-ogc-role'] = 'id' 

208 if k == p.time_field: 

209 queryables['properties'][k]['x-ogc-role'] = 'primary-instant' # noqa 

210 if domains.get(k): 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true

211 queryables['properties'][k]['enum'] = domains[k] 

212 

213 if request.format == F_HTML: # render 

214 tpl_config = api.get_dataset_templates(dataset) 

215 

216 queryables['title'] = l10n.translate( 

217 api.config['resources'][dataset]['title'], request.locale) 

218 

219 queryables['collections_path'] = api.get_collections_url() 

220 queryables['dataset_path'] = f'{api.get_collections_url()}/{dataset}' 

221 

222 content = render_j2_template(api.tpl_config, tpl_config, 

223 'collections/queryables.html', 

224 queryables, request.locale) 

225 

226 return headers, HTTPStatus.OK, content 

227 

228 headers['Content-Type'] = 'application/schema+json' 

229 

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

231 

232 

233def get_collection_items( 

234 api: API, request: Union[APIRequest, Any], 

235 dataset: str | None = None) -> Tuple[dict, int, str]: 

236 """ 

237 Queries collection 

238 

239 :param request: A request object 

240 :param dataset: dataset name 

241 

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

243 """ 

244 

245 # Set Content-Language to system locale until provider locale 

246 # has been determined 

247 headers = request.get_response_headers(SYSTEM_LOCALE, 

248 **api.api_headers) 

249 

250 properties = [] 

251 reserved_fieldnames = ['bbox', 'bbox-crs', 'crs', 'f', 'lang', 'limit', 

252 'offset', 'resulttype', 'datetime', 'sortby', 

253 'properties', 'skipGeometry', 'q', 

254 'filter', 'filter-lang', 'filter-crs'] 

255 

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

257 'type', 'collection') 

258 

259 if dataset not in collections.keys(): 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true

260 msg = 'Collection not found' 

261 return api.get_exception( 

262 HTTPStatus.NOT_FOUND, headers, request.format, 'NotFound', msg) 

263 

264 LOGGER.debug('Processing query parameters') 

265 

266 LOGGER.debug('Processing offset parameter') 

267 try: 

268 offset = int(request.params.get('offset')) 

269 if offset < 0: 

270 msg = 'offset value should be positive or zero' 

271 return api.get_exception( 

272 HTTPStatus.BAD_REQUEST, headers, request.format, 

273 'InvalidParameterValue', msg) 

274 except ValueError: 

275 msg = 'offset value should be an integer' 

276 return api.get_exception( 

277 HTTPStatus.BAD_REQUEST, headers, request.format, 

278 'InvalidParameterValue', msg) 

279 except TypeError as err: 

280 LOGGER.warning(err) 

281 offset = 0 

282 

283 LOGGER.debug('Processing limit parameter') 

284 if api.config['server'].get('limit') is not None: 284 ↛ 285line 284 didn't jump to line 285 because the condition on line 284 was never true

285 msg = ('server.limit is no longer supported! ' 

286 'Please use limits at the server or collection ' 

287 'level (RFC5)') 

288 LOGGER.warning(msg) 

289 try: 

290 limit = evaluate_limit(request.params.get('limit'), 

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

292 collections[dataset].get('limits', {})) 

293 except ValueError as err: 

294 return api.get_exception( 

295 HTTPStatus.BAD_REQUEST, headers, request.format, 

296 'InvalidParameterValue', str(err)) 

297 

298 resulttype = request.params.get('resulttype') or 'results' 

299 

300 LOGGER.debug('Processing bbox parameter') 

301 

302 bbox = request.params.get('bbox') 

303 

304 if bbox is None: 

305 bbox = [] 

306 else: 

307 try: 

308 bbox = validate_bbox(bbox) 

309 except ValueError as err: 

310 msg = str(err) 

311 return api.get_exception( 

312 HTTPStatus.BAD_REQUEST, headers, request.format, 

313 'InvalidParameterValue', msg) 

314 

315 LOGGER.debug('Processing datetime parameter') 

316 datetime_ = request.params.get('datetime') 

317 try: 

318 datetime_ = validate_datetime(collections[dataset]['extents'], 

319 datetime_) 

320 except ValueError as err: 

321 msg = str(err) 

322 return api.get_exception( 

323 HTTPStatus.BAD_REQUEST, headers, request.format, 

324 'InvalidParameterValue', msg) 

325 

326 LOGGER.debug('processing q parameter') 

327 q = request.params.get('q') or None 

328 

329 LOGGER.debug('Loading provider') 

330 

331 provider_def = None 

332 try: 

333 provider_type = 'feature' 

334 provider_def = get_provider_by_type( 

335 collections[dataset]['providers'], provider_type) 

336 p = load_plugin('provider', provider_def) 

337 except ProviderTypeError: 

338 try: 

339 provider_type = 'record' 

340 provider_def = get_provider_by_type( 

341 collections[dataset]['providers'], provider_type) 

342 p = load_plugin('provider', provider_def) 

343 except ProviderTypeError: 

344 msg = 'Invalid provider type' 

345 return api.get_exception( 

346 HTTPStatus.BAD_REQUEST, headers, request.format, 

347 'NoApplicableCode', msg) 

348 except ProviderGenericError as err: 

349 return api.get_exception( 

350 err.http_status_code, headers, request.format, 

351 err.ogc_exception_code, err.message) 

352 

353 LOGGER.debug('Validating requested format') 

354 dataset_formatters = get_dataset_formatters(collections[dataset]) 

355 

356 if dataset_formatters: 356 ↛ 364line 356 didn't jump to line 364 because the condition on line 356 was always true

357 LOGGER.debug(f'Dataset formatters: {dataset_formatters}') 

358 request._format = request._get_format( 

359 request.get_request_headers(request.headers), 

360 {v.f: v.mimetype for v in dataset_formatters.values()}) 

361 

362 LOGGER.debug(f'Request format: {request.format}') 

363 

364 if not request.is_valid(dataset_formatters.keys()): 

365 return api.get_format_exception(request) 

366 

367 crs_transform_spec = None 

368 if provider_type == 'feature': 368 ↛ 384line 368 didn't jump to line 384 because the condition on line 368 was always true

369 # crs query parameter is only available for OGC API - Features 

370 # right now, not for OGC API - Records. 

371 LOGGER.debug('Processing crs parameter') 

372 query_crs_uri = request.params.get('crs') 

373 try: 

374 crs_transform_spec = create_crs_transform_spec( 

375 provider_def, query_crs_uri 

376 ) 

377 except (ValueError, CRSError) as err: 

378 msg = str(err) 

379 return api.get_exception( 

380 HTTPStatus.BAD_REQUEST, headers, request.format, 

381 'InvalidParameterValue', msg) 

382 set_content_crs_header(headers, provider_def, query_crs_uri) 

383 

384 LOGGER.debug('Processing bbox-crs parameter') 

385 bbox_crs = request.params.get('bbox-crs') 

386 if bbox_crs is not None: 

387 # Validate bbox-crs parameter 

388 if len(bbox) == 0: 

389 msg = 'bbox-crs specified without bbox parameter' 

390 return api.get_exception( 

391 HTTPStatus.BAD_REQUEST, headers, request.format, 

392 'NoApplicableCode', msg) 

393 

394 if len(bbox_crs) == 0: 394 ↛ 395line 394 didn't jump to line 395 because the condition on line 394 was never true

395 msg = 'bbox-crs specified but is empty' 

396 return api.get_exception( 

397 HTTPStatus.BAD_REQUEST, headers, request.format, 

398 'NoApplicableCode', msg) 

399 

400 supported_crs_list = get_supported_crs_list(provider_def) 

401 if bbox_crs not in supported_crs_list: 401 ↛ 402line 401 didn't jump to line 402 because the condition on line 401 was never true

402 msg = f'bbox-crs {bbox_crs} not supported for this collection' 

403 return api.get_exception( 

404 HTTPStatus.BAD_REQUEST, headers, request.format, 

405 'NoApplicableCode', msg) 

406 elif len(bbox) > 0: 406 ↛ 408line 406 didn't jump to line 408 because the condition on line 406 was never true

407 # bbox but no bbox-crs param: assume bbox is in default CRS 

408 bbox_crs = DEFAULT_STORAGE_CRS 

409 

410 # Transform bbox to storage_crs 

411 # when bbox-crs different from storage_crs. 

412 if len(bbox) > 0: 

413 try: 

414 # Get a pyproj CRS instance for the Collection's Storage CRS 

415 storage_crs = provider_def.get('storage_crs', DEFAULT_STORAGE_CRS) 

416 

417 # Do the (optional) Transform to the Storage CRS 

418 bbox = transform_bbox(bbox, bbox_crs, storage_crs) 

419 except CRSError as e: 

420 return api.get_exception( 

421 HTTPStatus.BAD_REQUEST, headers, request.format, 

422 'NoApplicableCode', str(e)) 

423 

424 LOGGER.debug('processing property parameters') 

425 for k, v in request.params.items(): 

426 include_query_param = False 

427 if k not in reserved_fieldnames: 

428 if k in list(p.fields.keys()) or p.include_extra_query_parameters: 

429 include_query_param = True 

430 

431 if include_query_param: 

432 LOGGER.debug(f'Including query parameter {k}={v}') 

433 properties.append((k, v)) 

434 else: 

435 LOGGER.debug(f'Discarding query parameter {k}={v}') 

436 

437 LOGGER.debug('processing sort parameter') 

438 val = request.params.get('sortby') 

439 

440 if val is not None: 

441 sortby = [] 

442 sorts = val.split(',') 

443 for s in sorts: 443 ↛ 460line 443 didn't jump to line 460 because the loop on line 443 didn't complete

444 prop = s 

445 order = '+' 

446 if s and s[0] in ['+', '-']: 

447 order = s[0] 

448 prop = s[1:] 

449 

450 if prop not in p.fields.keys(): 450 ↛ 456line 450 didn't jump to line 456 because the condition on line 450 was always true

451 msg = 'bad sortby property' 

452 return api.get_exception( 

453 HTTPStatus.BAD_REQUEST, headers, request.format, 

454 'InvalidParameterValue', msg) 

455 

456 sortby.append({'property': prop, 'order': order}) 

457 else: 

458 sortby = [] 

459 

460 LOGGER.debug('processing properties parameter') 

461 val = request.params.get('properties') 

462 

463 if val is not None: 

464 select_properties = val.split(',') 

465 properties_to_check = set(p.properties) | set(p.fields.keys()) 

466 

467 if (len(list(set(select_properties) - 

468 set(properties_to_check))) > 0): 

469 msg = 'unknown properties specified' 

470 return api.get_exception( 

471 HTTPStatus.BAD_REQUEST, headers, request.format, 

472 'InvalidParameterValue', msg) 

473 else: 

474 select_properties = [] 

475 

476 LOGGER.debug('processing skipGeometry parameter') 

477 val = request.params.get('skipGeometry') 

478 if val is not None: 

479 skip_geometry = str2bool(val) 

480 else: 

481 skip_geometry = False 

482 

483 LOGGER.debug('Processing filter-crs parameter') 

484 filter_crs_uri = request.params.get('filter-crs', DEFAULT_CRS) 

485 

486 LOGGER.debug('processing filter parameter') 

487 cql_text = request.params.get('filter') 

488 

489 if cql_text is not None: 489 ↛ 490line 489 didn't jump to line 490 because the condition on line 489 was never true

490 try: 

491 filter_ = parse_ecql_text(cql_text) 

492 filter_ = modify_pygeofilter( 

493 filter_, 

494 filter_crs_uri=filter_crs_uri, 

495 storage_crs_uri=provider_def.get('storage_crs'), 

496 geometry_column_name=provider_def.get('geom_field'), 

497 ) 

498 except Exception: 

499 msg = 'Bad CQL text' 

500 LOGGER.error(f'{msg}: {cql_text}') 

501 return api.get_exception( 

502 HTTPStatus.BAD_REQUEST, headers, request.format, 

503 'InvalidParameterValue', msg) 

504 elif request.data: 

505 try: 

506 request_data = request.data.decode() 

507 filter_ = parse_cql2_json(request_data) 

508 filter_ = modify_pygeofilter( 

509 filter_, 

510 filter_crs_uri=filter_crs_uri, 

511 storage_crs_uri=provider_def.get('storage_crs'), 

512 geometry_column_name=provider_def.get('geom_field'), 

513 ) 

514 except Exception: 

515 msg = 'Bad CQL JSON' 

516 LOGGER.error(f'{msg}: {request_data}') 

517 return api.get_exception( 

518 HTTPStatus.BAD_REQUEST, headers, request.format, 

519 'InvalidParameterValue', msg) 

520 else: 

521 filter_ = None 

522 

523 LOGGER.debug('Processing filter-lang parameter') 

524 filter_lang = request.params.get('filter-lang') 

525 # Currently only cql-text is handled, but it is optional 

526 if filter_lang not in [None, 'cql-json', 'cql-text']: 526 ↛ 527line 526 didn't jump to line 527 because the condition on line 526 was never true

527 msg = 'Invalid filter language' 

528 return api.get_exception( 

529 HTTPStatus.BAD_REQUEST, headers, request.format, 

530 'InvalidParameterValue', msg) 

531 # Get provider locale (if any) 

532 prv_locale = l10n.get_plugin_locale(provider_def, request.raw_locale) 

533 

534 LOGGER.debug('Querying provider') 

535 LOGGER.debug(f'offset: {offset}') 

536 LOGGER.debug(f'limit: {limit}') 

537 LOGGER.debug(f'resulttype: {resulttype}') 

538 LOGGER.debug(f'sortby: {sortby}') 

539 LOGGER.debug(f'bbox: {bbox}') 

540 if provider_type == 'feature': 540 ↛ 542line 540 didn't jump to line 542 because the condition on line 540 was always true

541 LOGGER.debug(f'crs: {query_crs_uri}') 

542 LOGGER.debug(f'datetime: {datetime_}') 

543 LOGGER.debug(f'properties: {properties}') 

544 LOGGER.debug(f'select properties: {select_properties}') 

545 LOGGER.debug(f'skipGeometry: {skip_geometry}') 

546 LOGGER.debug(f'language: {prv_locale}') 

547 LOGGER.debug(f'q: {q}') 

548 LOGGER.debug(f'cql_text: {cql_text}') 

549 LOGGER.debug(f'filter_: {filter_}') 

550 LOGGER.debug(f'filter-lang: {filter_lang}') 

551 LOGGER.debug(f'filter-crs: {filter_crs_uri}') 

552 

553 try: 

554 content = p.query(offset=offset, limit=limit, 

555 resulttype=resulttype, bbox=bbox, 

556 datetime_=datetime_, properties=properties, 

557 sortby=sortby, skip_geometry=skip_geometry, 

558 select_properties=select_properties, 

559 crs_transform_spec=crs_transform_spec, 

560 q=q, language=prv_locale, filterq=filter_) 

561 except ProviderGenericError as err: 

562 return api.get_exception( 

563 err.http_status_code, headers, request.format, 

564 err.ogc_exception_code, err.message) 

565 

566 serialized_query_params = '' 

567 for k, v in request.params.items(): 

568 if k not in ('f', 'offset'): 

569 serialized_query_params += '&' 

570 serialized_query_params += urllib.parse.quote(k, safe='') 

571 serialized_query_params += '=' 

572 serialized_query_params += urllib.parse.quote(str(v), safe=',') 

573 

574 if 'links' not in content: 574 ↛ 578line 574 didn't jump to line 578 because the condition on line 574 was always true

575 content['links'] = [] 

576 

577 # TODO: translate titles 

578 uri = f'{api.get_collections_url()}/{dataset}/items' 

579 content['links'].extend([{ 

580 'type': 'application/geo+json', 

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

582 'title': l10n.translate('This document as GeoJSON', request.locale), 

583 'href': f'{uri}?f={F_JSON}{serialized_query_params}' 

584 }, { 

585 'rel': request.get_linkrel(F_JSONLD), 

586 'type': FORMAT_TYPES[F_JSONLD], 

587 'title': l10n.translate('This document as RDF (JSON-LD)', request.locale), # noqa 

588 'href': f'{uri}?f={F_JSONLD}{serialized_query_params}' 

589 }, { 

590 'type': FORMAT_TYPES[F_HTML], 

591 'rel': request.get_linkrel(F_HTML), 

592 'title': l10n.translate('This document as HTML', request.locale), 

593 'href': f'{uri}?f={F_HTML}{serialized_query_params}' 

594 }]) 

595 

596 for key, value in dataset_formatters.items(): 

597 content['links'].append({ 

598 'type': value.mimetype, 

599 'rel': 'alternate', 

600 'title': f'This document as {key}', 

601 'href': f'{uri}?f={value.name}{serialized_query_params}' 

602 }) 

603 

604 next_link = False 

605 prev_link = False 

606 

607 if 'next' in [link['rel'] for link in content['links']]: 607 ↛ 608line 607 didn't jump to line 608 because the condition on line 607 was never true

608 LOGGER.debug('Using next link from provider') 

609 else: 

610 if content.get('numberMatched', -1) > (limit + offset): 

611 next_link = True 

612 elif len(content['features']) == limit: 

613 next_link = True 

614 

615 if offset > 0: 

616 prev_link = True 

617 

618 print(request.format) 

619 if prev_link: 

620 prev = max(0, offset - limit) 

621 url = f'{uri}?offset={prev}{serialized_query_params}' 

622 if request.format is not None: 

623 url = f'{uri}?f={request.format}&offset={prev}{serialized_query_params}' # noqa 

624 

625 content['links'].append( 

626 { 

627 'type': 'application/geo+json', 

628 'rel': 'prev', 

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

630 'href': url 

631 }) 

632 

633 if next_link: 

634 next_ = offset + limit 

635 next_href = f'{uri}?offset={next_}{serialized_query_params}' 

636 if request.format is not None: 

637 next_href = f'{uri}?f={request.format}&offset={next_}{serialized_query_params}' # noqa 

638 content['links'].append( 

639 { 

640 'type': 'application/geo+json', 

641 'rel': 'next', 

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

643 'href': next_href 

644 }) 

645 

646 content['links'].append( 

647 { 

648 'type': FORMAT_TYPES[F_JSON], 

649 'title': l10n.translate( 

650 collections[dataset]['title'], request.locale), 

651 'rel': 'collection', 

652 'href': '/'.join(uri.split('/')[:-1]) 

653 }) 

654 

655 content['timeStamp'] = datetime.now(UTC).strftime( 

656 '%Y-%m-%dT%H:%M:%S.%fZ') 

657 

658 # Set response language to requested provider locale 

659 # (if it supports language) and/or otherwise the requested pygeoapi 

660 # locale (or fallback default locale) 

661 l10n.set_response_language(headers, prv_locale, request.locale) 

662 

663 if request.format == F_HTML: # render 

664 tpl_config = api.get_dataset_templates(dataset) 

665 # For constructing proper URIs to items 

666 

667 content['itemtype'] = p.type 

668 content['items_path'] = uri 

669 content['dataset_path'] = '/'.join(uri.split('/')[:-1]) 

670 content['collections_path'] = api.get_collections_url() 

671 

672 content['offset'] = offset 

673 

674 content['id_field'] = p.id_field 

675 if p.uri_field is not None: 675 ↛ 676line 675 didn't jump to line 676 because the condition on line 675 was never true

676 content['uri_field'] = p.uri_field 

677 if p.title_field is not None: 677 ↛ 678line 677 didn't jump to line 678 because the condition on line 677 was never true

678 content['title_field'] = l10n.translate(p.title_field, 

679 request.locale) 

680 # If title exists, use it as id in html templates 

681 content['id_field'] = content['title_field'] 

682 content = render_j2_template(api.tpl_config, tpl_config, 

683 'collections/items/index.html', 

684 content, request.locale) 

685 return headers, HTTPStatus.OK, content 

686 elif request.format in [df.f for df in dataset_formatters.values()]: 

687 formatter = [v for v in dataset_formatters.values() if 

688 v.f == request.format][0] 

689 

690 try: 

691 content = formatter.write( 

692 data=content, 

693 options={ 

694 'provider_def': get_provider_by_type( 

695 collections[dataset]['providers'], 

696 'feature') 

697 } 

698 ) 

699 except FormatterSerializationError: 

700 msg = 'Error serializing output' 

701 return api.get_exception( 

702 HTTPStatus.INTERNAL_SERVER_ERROR, headers, request.format, 

703 'NoApplicableCode', msg) 

704 

705 headers['Content-Type'] = formatter.mimetype 

706 

707 if formatter.attachment: 707 ↛ 708line 707 didn't jump to line 708 because the condition on line 707 was never true

708 if p.filename is None: 

709 filename = f'{dataset}.{formatter.extension}' 

710 else: 

711 filename = f'{p.filename}' 

712 

713 cd = f'attachment; filename="{filename}"' 

714 headers['Content-Disposition'] = cd 

715 

716 return headers, HTTPStatus.OK, content 

717 

718 elif request.format == F_JSONLD: 

719 content = geojson2jsonld( 

720 api, content, dataset, id_field=(p.uri_field or 'id') 

721 ) 

722 

723 return headers, HTTPStatus.OK, content 

724 

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

726 

727 

728def manage_collection_item( 

729 api: API, request: APIRequest, 

730 action: str, dataset: str, 

731 identifier: str | None = None) -> Tuple[dict, int, str]: 

732 """ 

733 Adds an item to a collection 

734 

735 :param request: A request object 

736 :param action: an action among 'create', 'update', 'delete', 'options' 

737 :param dataset: dataset name 

738 

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

740 """ 

741 

742 if not request.is_valid(PLUGINS['formatter'].keys()): 742 ↛ 743line 742 didn't jump to line 743 because the condition on line 742 was never true

743 return api.get_format_exception(request) 

744 

745 # Set Content-Language to system locale until provider locale 

746 # has been determined 

747 headers = request.get_response_headers(SYSTEM_LOCALE, **api.api_headers) 

748 

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

750 'type', 'collection') 

751 

752 http_status = HTTPStatus.OK 

753 payload = None 

754 

755 if dataset not in collections.keys(): 755 ↛ 756line 755 didn't jump to line 756 because the condition on line 755 was never true

756 msg = 'Collection not found' 

757 return api.get_exception( 

758 HTTPStatus.NOT_FOUND, headers, request.format, 'NotFound', msg) 

759 

760 LOGGER.debug('Loading provider') 

761 try: 

762 provider_def = get_provider_by_type( 

763 collections[dataset]['providers'], 'feature') 

764 p = load_plugin('provider', provider_def) 

765 except ProviderTypeError: 

766 try: 

767 provider_def = get_provider_by_type(