huanld/pygeoapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Proposal to split api.py into different files (#1405)

Browse Source 
* Move api to subdirectory

* Move processes api to own file

* Adapt processes view methods

* Move openapi definition to processes api

* Use processes api in flask

* Linter

* Fix import issues

* Allow calling refactored views from starlette

* Allow calling refactored views from django

* Linter

* Move edr api to own file

* Adapt edr api to new style

* Fix typo in django views

* Move maps api to own file

* Adapt maps api to new style

* Move edr openapi to edr api file

* Move maps openapi to maps api file

* Move stac views to own file

* Refactor stac views to new file

* Move stac openapi to stac api file

* Move tiles api to own file

* Adapt tiles api to new style

* Also move tilematrixset to tiles api

* Adapt tilesetmatrix views to new style

NOTE: I had to remove one tilematrixsets test because
it tested that an invalid format would produce an error.
This now happens by default for all views, but the actual
code is outside of the endpoint function.

* update features, records, coverages

* update release version

* switch back to dev

* backport of #1313

* backport of #1313 fix

* backport of #1585

* Flask: sanitize OGC schema pathing (#1593)

* update release version

* switch back to dev

* backport of #1596

* Port test_gzip_csv test

Note that apply_gzip is now called by the web framework adapters,
so to test it in general, we have to call it in the test manually

* Add empty conformance class list to stac api

* Fix queryables call in starlette

* fix ref

* Unify request validity checking

The default case is handled by the web framework adapters. If custom
format handling is required, the check in the adapter must be skipped.

* Fix imports in django views

* backport #1598

* Remove test about format handling in endpoint

This is now handled outside of the endpoint function

* add docstring to base process manager (#1603)

* backport of #1601

* Port api ogr tests to new style

* Move processes tests to own file

* Run api tests from new dir in CI

* Move edr tests to own file

* Move maps tests to own file

* Move tiles tests to own file

* Actually hide hidden layers in openapi

* 1600 allow providing default value in config (#1604)

* move coverages tests to own file

* move itemtypes to own file, move core into init test

* fix OpenAPI output

* update tests

* add missing descriptions to OpenAPI admin responses

* update tests

* fix tests autodiscovery

* remove unused logging in tests

* address PR comments

* test with xarray 2024.2.0

* remove unneeded file

* safeguard xarray error

* unpin xarray

* fix OpenAPI generation

* fix schema endpoint in Flask and Starlette

* Safely serialize configuration JSON (#1605)

* Safely serialize configuration JSON

Co-Authored-By: Tom Kralidis <tomkralidis@gmail.com>

* Revert "Safely serialize configuration JSON"

This reverts commit 36feb067ee6f87e61955852dc48994f075806370.

* Add test for datetime with Admin API

* Safely serialize configuration JSON

---------

Co-authored-by: Tom Kralidis <tomkralidis@gmail.com>

* backport #1611

* Also fix schema endpoint for django

Fix is analogous to e72d4ba3a5ba3b8621ca839e7814429beeeb8f01

* address additional PR comments

---------

Co-authored-by: Tom Kralidis <tomkralidis@gmail.com>
Co-authored-by: Angelos Tzotsos <gcpp.kalxas@gmail.com>
Co-authored-by: Ricardo Garcia Silva <ricardo.garcia.silva@gmail.com>
Co-authored-by: Benjamin Webb <40066515+webb-ben@users.noreply.github.com>
This commit is contained in:
Bernhard Mallinger
2024-04-05 12:06:25 +02:00
committed by GitHub
parent 36e75f8b72
commit 35bdcb6f02
31 changed files with 8918 additions and 7812 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pygeoapi/api/__init__.py
+1687
View File
File diff suppressed because it is too large Load Diff
pygeoapi/api/coverages.py
+253
View File
@@ -0,0 +1,253 @@
# =================================================================
# Authors: Tom Kralidis <tomkralidis@gmail.com>
# Francesco Bartoli <xbartolone@gmail.com>
# Sander Schaminee <sander.schaminee@geocat.net>
# John A Stevenson <jostev@bgs.ac.uk>
# Colin Blackburn <colb@bgs.ac.uk>
# Ricardo Garcia Silva <ricardo.garcia.silva@geobeyond.it>
# Bernhard Mallinger <bernhard.mallinger@eox.at>
#
# Copyright (c) 2024 Tom Kralidis
# Copyright (c) 2022 Francesco Bartoli
# Copyright (c) 2022 John A Stevenson and Colin Blackburn
# Copyright (c) 2023 Ricardo Garcia Silva
# Copyright (c) 2024 Bernhard Mallinger
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without
# restriction, including without limitation the rights to use,
# copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following
# conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
#
# =================================================================
import logging
from http import HTTPStatus
from typing import Tuple
from pygeoapi import l10n
from pygeoapi.plugin import load_plugin
from pygeoapi.provider.base import ProviderGenericError, ProviderTypeError
from pygeoapi.util import (
filter_dict_by_key_value, get_provider_by_type, to_json
)
from . import (
APIRequest, API, F_JSON, SYSTEM_LOCALE, validate_bbox, validate_datetime,
validate_subset
)
LOGGER = logging.getLogger(__name__)
CONFORMANCE_CLASSES = [
'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/core',
'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/oas30',
'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/html',
'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/geodata-coverage',
'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/coverage-subset',
'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/coverage-rangesubset', # noqa
'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/coverage-bbox',
'http://www.opengis.net/spec/ogcapi-coverages-1/1.0/conf/coverage-datetime'
]
def get_collection_coverage(
api: API, request: APIRequest, dataset) -> Tuple[dict, int, str]:
"""
Returns a subset of a collection coverage
:param request: A request object
:param dataset: dataset name
:returns: tuple of headers, status code, content
"""
query_args = {}
format_ = request.format or F_JSON
# Force response content type and language (en-US only) headers
headers = request.get_response_headers(SYSTEM_LOCALE, **api.api_headers)
LOGGER.debug('Loading provider')
try:
collection_def = get_provider_by_type(
api.config['resources'][dataset]['providers'], 'coverage')
p = load_plugin('provider', collection_def)
except KeyError:
msg = 'collection does not exist'
return api.get_exception(
HTTPStatus.NOT_FOUND, headers, format_,
'InvalidParameterValue', msg)
except ProviderGenericError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers, request.format,
err.ogc_exception_code, err.message)
LOGGER.debug('Processing bbox parameter')
bbox = request.params.get('bbox')
if bbox is None:
bbox = []
else:
try:
bbox = validate_bbox(bbox)
except ValueError as err:
msg = str(err)
return api.get_exception(
HTTPStatus.INTERNAL_SERVER_ERROR, headers, format_,
'InvalidParameterValue', msg)
query_args['bbox'] = bbox
LOGGER.debug('Processing bbox-crs parameter')
bbox_crs = request.params.get('bbox-crs')
if bbox_crs is not None:
query_args['bbox_crs'] = bbox_crs
LOGGER.debug('Processing datetime parameter')
datetime_ = request.params.get('datetime')
try:
datetime_ = validate_datetime(
api.config['resources'][dataset]['extents'], datetime_)
except ValueError as err:
msg = str(err)
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, format_,
'InvalidParameterValue', msg)
query_args['datetime_'] = datetime_
query_args['format_'] = format_
properties = request.params.get('properties')
if properties:
LOGGER.debug('Processing properties parameter')
query_args['properties'] = [rs for
rs in properties.split(',') if rs]
LOGGER.debug(f"Fields: {query_args['properties']}")
for a in query_args['properties']:
if a not in p.fields:
msg = 'Invalid field specified'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, format_,
'InvalidParameterValue', msg)
if 'subset' in request.params:
LOGGER.debug('Processing subset parameter')
try:
subsets = validate_subset(request.params['subset'] or '')
except (AttributeError, ValueError) as err:
msg = f'Invalid subset: {err}'
LOGGER.error(msg)
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, format_,
'InvalidParameterValue', msg)
if not set(subsets.keys()).issubset(p.axes):
msg = 'Invalid axis name'
LOGGER.error(msg)
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, format_,
'InvalidParameterValue', msg)
query_args['subsets'] = subsets
LOGGER.debug(f"Subsets: {query_args['subsets']}")
LOGGER.debug('Querying coverage')
try:
data = p.query(**query_args)
except ProviderGenericError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers, request.format,
err.ogc_exception_code, err.message)
mt = collection_def['format']['name']
if format_ == mt: # native format
if p.filename is not None:
cd = f'attachment; filename="{p.filename}"'
headers['Content-Disposition'] = cd
headers['Content-Type'] = collection_def['format']['mimetype']
return headers, HTTPStatus.OK, data
elif format_ == F_JSON:
headers['Content-Type'] = 'application/prs.coverage+json'
return headers, HTTPStatus.OK, to_json(data, api.pretty_print)
else:
return api.get_format_exception(request)
def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str, dict]]: # noqa
"""
Get OpenAPI fragments
:param cfg: `dict` of configuration
:param locale: `str` of locale
:returns: `tuple` of `list` of tag objects, and `dict` of path objects
"""
from pygeoapi.openapi import OPENAPI_YAML, get_visible_collections
paths = {}
collections = filter_dict_by_key_value(cfg['resources'],
'type', 'collection')
for k, v in get_visible_collections(cfg).items():
try:
load_plugin('provider', get_provider_by_type(
collections[k]['providers'], 'coverage'))
except ProviderTypeError:
LOGGER.debug('collection is not coverage based')
continue
coverage_path = f'/collections/{k}/coverage'
title = l10n.translate(v['title'], locale)
description = l10n.translate(v['description'], locale)
paths[coverage_path] = {
'get': {
'summary': f'Get {title} coverage',
'description': description,
'tags': [k],
'operationId': f'get{k.capitalize()}Coverage',
'parameters': [
{'$ref': '#/components/parameters/lang'},
{'$ref': '#/components/parameters/f'},
{'$ref': '#/components/parameters/bbox'},
{'$ref': '#/components/parameters/bbox-crs'}
],
'responses': {
'200': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/Features"}, # noqa
'400': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/InvalidParameter"}, # noqa
'404': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/NotFound"}, # noqa
'500': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/ServerError"} # noqa
}
}
}
return [{'name': 'coverages'}], {'paths': paths}
pygeoapi/api/environmental_data_retrieval.py
+332
View File
@@ -0,0 +1,332 @@
# =================================================================
# Authors: Tom Kralidis <tomkralidis@gmail.com>
# Francesco Bartoli <xbartolone@gmail.com>
# Sander Schaminee <sander.schaminee@geocat.net>
# John A Stevenson <jostev@bgs.ac.uk>
# Colin Blackburn <colb@bgs.ac.uk>
# Ricardo Garcia Silva <ricardo.garcia.silva@geobeyond.it>
# Bernhard Mallinger <bernhard.mallinger@eox.at>
#
# Copyright (c) 2024 Tom Kralidis
# Copyright (c) 2022 Francesco Bartoli
# Copyright (c) 2022 John A Stevenson and Colin Blackburn
# Copyright (c) 2023 Ricardo Garcia Silva
# Copyright (c) 2024 Bernhard Mallinger
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without
# restriction, including without limitation the rights to use,
# copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following
# conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
#
# =================================================================
from http import HTTPStatus
import logging
from typing import Tuple
from shapely.errors import WKTReadingError
from shapely.wkt import loads as shapely_loads
from pygeoapi.plugin import load_plugin, PLUGINS
from pygeoapi.provider.base import ProviderGenericError
from pygeoapi.util import (
filter_providers_by_type, get_provider_by_type, render_j2_template,
to_json, filter_dict_by_key_value
)
from . import APIRequest, API, F_HTML, validate_datetime, validate_bbox
LOGGER = logging.getLogger(__name__)
CONFORMANCE_CLASSES = [
'http://www.opengis.net/spec/ogcapi-edr-1/1.0/conf/core'
]
def get_collection_edr_query(api: API, request: APIRequest,
dataset, instance, query_type,
location_id=None) -> Tuple[dict, int, str]:
"""
Queries collection EDR
:param request: APIRequest instance with query params
:param dataset: dataset name
:param instance: instance name
:param query_type: EDR query type
:param location_id: location id of a /location/<location_id> query
:returns: tuple of headers, status code, content
"""
if not request.is_valid(PLUGINS['formatter'].keys()):
return api.get_format_exception(request)
headers = request.get_response_headers(api.default_locale,
**api.api_headers)
collections = filter_dict_by_key_value(api.config['resources'],
'type', 'collection')
if dataset not in collections.keys():
msg = 'Collection not found'
return api.get_exception(
HTTPStatus.NOT_FOUND, headers, request.format, 'NotFound', msg)
LOGGER.debug('Processing query parameters')
LOGGER.debug('Processing datetime parameter')
datetime_ = request.params.get('datetime')
try:
datetime_ = validate_datetime(collections[dataset]['extents'],
datetime_)
except ValueError as err:
msg = str(err)
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
LOGGER.debug('Processing parameter_names parameter')
parameternames = request.params.get('parameter_names') or []
if isinstance(parameternames, str):
parameternames = parameternames.split(',')
bbox = None
if query_type in ['cube', 'locations']:
LOGGER.debug('Processing cube bbox')
try:
bbox = validate_bbox(request.params.get('bbox'))
if not bbox and query_type == 'cube':
raise ValueError('bbox parameter required by cube queries')
except ValueError as err:
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', str(err))
LOGGER.debug('Processing coords parameter')
wkt = request.params.get('coords')
if wkt:
try:
wkt = shapely_loads(wkt)
except WKTReadingError:
msg = 'invalid coords parameter'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
elif query_type not in ['cube', 'locations']:
msg = 'missing coords parameter'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
within = within_units = None
if query_type == 'radius':
LOGGER.debug('Processing within / within-units parameters')
within = request.params.get('within')
within_units = request.params.get('within-units')
LOGGER.debug('Processing z parameter')
z = request.params.get('z')
LOGGER.debug('Loading provider')
try:
p = load_plugin('provider', get_provider_by_type(
collections[dataset]['providers'], 'edr'))
except ProviderGenericError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers, request.format,
err.ogc_exception_code, err.message)
if instance is not None and not p.get_instance(instance):
msg = 'Invalid instance identifier'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers,
request.format, 'InvalidParameterValue', msg)
if query_type not in p.get_query_types():
msg = 'Unsupported query type'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
if parameternames and not any((fld in parameternames)
for fld in p.get_fields().keys()):
msg = 'Invalid parameter_names'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
query_args = dict(
query_type=query_type,
instance=instance,
format_=request.format,
datetime_=datetime_,
select_properties=parameternames,
wkt=wkt,
z=z,
bbox=bbox,
within=within,
within_units=within_units,
limit=int(api.config['server']['limit']),
location_id=location_id,
)
try:
data = p.query(**query_args)
except ProviderGenericError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers, request.format,
err.ogc_exception_code, err.message)
if request.format == F_HTML: # render
content = render_j2_template(api.tpl_config,
'collections/edr/query.html', data,
api.default_locale)
else:
content = to_json(data, api.pretty_print)
return headers, HTTPStatus.OK, content
def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str, dict]]: # noqa
"""
Get OpenAPI fragments
:param cfg: `dict` of configuration
:param locale: `str` of locale
:returns: `tuple` of `list` of tag objects, and `dict` of path objects
"""
from pygeoapi.openapi import OPENAPI_YAML, get_visible_collections
LOGGER.debug('setting up edr endpoints')
paths = {}
collections = filter_dict_by_key_value(cfg['resources'],
'type', 'collection')
for k, v in get_visible_collections(cfg).items():
edr_extension = filter_providers_by_type(
collections[k]['providers'], 'edr')
if edr_extension:
collection_name_path = f'/collections/{k}'
ep = load_plugin('provider', edr_extension)
edr_query_endpoints = []
for qt in [qt for qt in ep.get_query_types() if qt != 'locations']:
edr_query_endpoints.append({
'path': f'{collection_name_path}/{qt}',
'qt': qt,
'op_id': f'query{qt.capitalize()}{k.capitalize()}'
})
if ep.instances:
edr_query_endpoints.append({
'path': f'{collection_name_path}/instances/{{instanceId}}/{qt}', # noqa
'qt': qt,
'op_id': f'query{qt.capitalize()}Instance{k.capitalize()}' # noqa
})
for eqe in edr_query_endpoints:
if eqe['qt'] == 'cube':
spatial_parameter = 'bbox'
else:
spatial_parameter = f"{eqe['qt']}Coords"
paths[eqe['path']] = {
'get': {
'summary': f"query {v['description']} by {eqe['qt']}",
'description': v['description'],
'tags': [k],
'operationId': eqe['op_id'],
'parameters': [
{'$ref': f"{OPENAPI_YAML['oaedr']}/parameters/{spatial_parameter}.yaml"}, # noqa
{'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/parameters/datetime"}, # noqa
{'$ref': f"{OPENAPI_YAML['oaedr']}/parameters/parameter-name.yaml"}, # noqa
{'$ref': f"{OPENAPI_YAML['oaedr']}/parameters/z.yaml"}, # noqa
{'$ref': '#/components/parameters/f'}
],
'responses': {
'200': {
'description': 'Response',
'content': {
'application/prs.coverage+json': {
'schema': {
'$ref': f"{OPENAPI_YAML['oaedr']}/schemas/coverageJSON.yaml" # noqa
}
}
}
}
}
}
}
if 'locations' in ep.get_query_types():
paths[f'{collection_name_path}/locations'] = {
'get': {
'summary': f"Get pre-defined locations of {v['description']}", # noqa
'description': v['description'],
'tags': [k],
'operationId': f'queryLOCATIONS{k.capitalize()}',
'parameters': [
{'$ref': f"{OPENAPI_YAML['oaedr']}/parameters/bbox.yaml"}, # noqa
{'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/parameters/datetime"}, # noqa
{'$ref': '#/components/parameters/f'}
],
'responses': {
'200': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/Features"}, # noqa
'400': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/InvalidParameter"}, # noqa
'500': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/ServerError"} # noqa
}
}
}
paths[f'{collection_name_path}/locations/{{locId}}'] = {
'get': {
'summary': f"query {v['description']} by location",
'description': v['description'],
'tags': [k],
'operationId': f'queryLOCATIONSBYID{k.capitalize()}',
'parameters': [
{'$ref': f"{OPENAPI_YAML['oaedr']}/parameters/{spatial_parameter}.yaml"}, # noqa
{'$ref': f"{OPENAPI_YAML['oaedr']}/parameters/locationId.yaml"}, # noqa
{'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/parameters/datetime"}, # noqa
{'$ref': f"{OPENAPI_YAML['oaedr']}/parameters/parameter-name.yaml"}, # noqa
{'$ref': f"{OPENAPI_YAML['oaedr']}/parameters/z.yaml"}, # noqa
{'$ref': '#/components/parameters/f'}
],
'responses': {
'200': {
'description': 'Response',
'content': {
'application/prs.coverage+json': {
'schema': {
'$ref': f"{OPENAPI_YAML['oaedr']}/schemas/coverageJSON.yaml" # noqa
}
}
}
}
}
}
}
return [{'name': 'edr'}], {'paths': paths}
pygeoapi/api/itemtypes.py
+1610
View File
File diff suppressed because it is too large Load Diff
pygeoapi/api/maps.py
+340
View File
@@ -0,0 +1,340 @@
# =================================================================
# Authors: Tom Kralidis <tomkralidis@gmail.com>
# Francesco Bartoli <xbartolone@gmail.com>
# Sander Schaminee <sander.schaminee@geocat.net>
# John A Stevenson <jostev@bgs.ac.uk>
# Colin Blackburn <colb@bgs.ac.uk>
# Ricardo Garcia Silva <ricardo.garcia.silva@geobeyond.it>
# Bernhard Mallinger <bernhard.mallinger@eox.at>
#
# Copyright (c) 2024 Tom Kralidis
# Copyright (c) 2022 Francesco Bartoli
# Copyright (c) 2022 John A Stevenson and Colin Blackburn
# Copyright (c) 2023 Ricardo Garcia Silva
# Copyright (c) 2024 Bernhard Mallinger
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without
# restriction, including without limitation the rights to use,
# copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following
# conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
#
# =================================================================
from copy import deepcopy
from http import HTTPStatus
import logging
from typing import Tuple
from pygeoapi.openapi import get_oas_30_parameters
from pygeoapi.plugin import load_plugin
from pygeoapi.provider.base import ProviderGenericError
from pygeoapi.util import (
get_provider_by_type, to_json, filter_providers_by_type,
filter_dict_by_key_value
)
from . import APIRequest, API, validate_datetime
LOGGER = logging.getLogger(__name__)
CONFORMANCE_CLASSES = [
'http://www.opengis.net/spec/ogcapi-maps-1/1.0/conf/core'
]
def get_collection_map(api: API, request: APIRequest,
dataset, style=None) -> Tuple[dict, int, str]:
"""
Returns a subset of a collection map
:param request: A request object
:param dataset: dataset name
:param style: style name
:returns: tuple of headers, status code, content
"""
query_args = {
'crs': 'CRS84'
}
format_ = request.format or 'png'
headers = request.get_response_headers(**api.api_headers)
LOGGER.debug('Processing query parameters')
LOGGER.debug('Loading provider')
try:
collection_def = get_provider_by_type(
api.config['resources'][dataset]['providers'], 'map')
p = load_plugin('provider', collection_def)
except KeyError:
exception = {
'code': 'InvalidParameterValue',
'description': 'collection does not exist'
}
headers['Content-type'] = 'application/json'
LOGGER.error(exception)
return headers, HTTPStatus.NOT_FOUND, to_json(
exception, api.pretty_print)
except ProviderGenericError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers, request.format,
err.ogc_exception_code, err.message)
query_args['format_'] = request.params.get('f', 'png')
query_args['style'] = style
query_args['crs'] = request.params.get('bbox-crs', 4326)
query_args['transparent'] = request.params.get('transparent', True)
try:
query_args['width'] = int(request.params.get('width', 500))
query_args['height'] = int(request.params.get('height', 300))
except ValueError:
exception = {
'code': 'InvalidParameterValue',
'description': 'invalid width/height'
}
headers['Content-type'] = 'application/json'
LOGGER.error(exception)
return headers, HTTPStatus.BAD_REQUEST, to_json(
exception, api.pretty_print)
LOGGER.debug('Processing bbox parameter')
try:
bbox = request.params.get('bbox').split(',')
if len(bbox) != 4:
exception = {
'code': 'InvalidParameterValue',
'description': 'bbox values should be minx,miny,maxx,maxy'
}
headers['Content-type'] = 'application/json'
LOGGER.error(exception)
return headers, HTTPStatus.BAD_REQUEST, to_json(
exception, api.pretty_print)
except AttributeError:
bbox = api.config['resources'][dataset]['extents']['spatial']['bbox'] # noqa
try:
query_args['bbox'] = [float(c) for c in bbox]
except ValueError:
exception = {
'code': 'InvalidParameterValue',
'description': 'bbox values must be numbers'
}
headers['Content-type'] = 'application/json'
LOGGER.error(exception)
return headers, HTTPStatus.BAD_REQUEST, to_json(
exception, api.pretty_print)
LOGGER.debug('Processing datetime parameter')
datetime_ = request.params.get('datetime')
try:
query_args['datetime_'] = validate_datetime(
api.config['resources'][dataset]['extents'], datetime_)
except ValueError as err:
msg = str(err)
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
LOGGER.debug('Generating map')
try:
data = p.query(**query_args)
except ProviderGenericError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers, request.format,
err.ogc_exception_code, err.message)
mt = collection_def['format']['name']
if format_ == mt:
headers['Content-Type'] = collection_def['format']['mimetype']
return headers, HTTPStatus.OK, data
elif format_ in [None, 'html']:
headers['Content-Type'] = collection_def['format']['mimetype']
return headers, HTTPStatus.OK, data
else:
exception = {
'code': 'InvalidParameterValue',
'description': 'invalid format parameter'
}
LOGGER.error(exception)
return headers, HTTPStatus.BAD_REQUEST, to_json(
data, api.pretty_print)
def get_collection_map_legend(api: API, request: APIRequest,
dataset, style=None) -> Tuple[dict, int, str]:
"""
Returns a subset of a collection map legend
:param request: A request object
:param dataset: dataset name
:param style: style name
:returns: tuple of headers, status code, content
"""
format_ = 'png'
headers = request.get_response_headers(**api.api_headers)
LOGGER.debug('Processing query parameters')
LOGGER.debug('Loading provider')
try:
collection_def = get_provider_by_type(
api.config['resources'][dataset]['providers'], 'map')
p = load_plugin('provider', collection_def)
except KeyError:
exception = {
'code': 'InvalidParameterValue',
'description': 'collection does not exist'
}
LOGGER.error(exception)
return headers, HTTPStatus.NOT_FOUND, to_json(
exception, api.pretty_print)
except ProviderGenericError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers, request.format,
err.ogc_exception_code, err.message)
LOGGER.debug('Generating legend')
try:
data = p.get_legend(style, request.params.get('f', 'png'))
except ProviderGenericError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers, request.format,
err.ogc_exception_code, err.message)
mt = collection_def['format']['name']
if format_ == mt:
headers['Content-Type'] = collection_def['format']['mimetype']
return headers, HTTPStatus.OK, data
else:
exception = {
'code': 'InvalidParameterValue',
'description': 'invalid format parameter'
}
LOGGER.error(exception)
return headers, HTTPStatus.BAD_REQUEST, to_json(
data, api.pretty_print)
def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str, dict]]: # noqa
"""
Get OpenAPI fragments
:param cfg: `dict` of configuration
:param locale: `str` of locale
:returns: `tuple` of `list` of tag objects, and `dict` of path objects
"""
from pygeoapi.openapi import OPENAPI_YAML, get_visible_collections
LOGGER.debug('setting up maps endpoints')
paths = {}
collections = filter_dict_by_key_value(cfg['resources'],
'type', 'collection')
parameters = get_oas_30_parameters(cfg, locale)
for k, v in get_visible_collections(cfg).items():
map_extension = filter_providers_by_type(
collections[k]['providers'], 'map')
if map_extension:
mp = load_plugin('provider', map_extension)
map_f = deepcopy(parameters['f'])
map_f['schema']['enum'] = [map_extension['format']['name']]
map_f['schema']['default'] = map_extension['format']['name']
pth = f'/collections/{k}/map'
paths[pth] = {
'get': {
'summary': 'Get map',
'description': f"{v['description']} map",
'tags': [k],
'operationId': 'getMap',
'parameters': [
{'$ref': '#/components/parameters/bbox'},
{'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/parameters/datetime"}, # noqa
{
'name': 'width',
'in': 'query',
'description': 'Response image width',
'required': False,
'schema': {
'type': 'integer',
},
'style': 'form',
'explode': False
},
{
'name': 'height',
'in': 'query',
'description': 'Response image height',
'required': False,
'schema': {
'type': 'integer',
},
'style': 'form',
'explode': False
},
{
'name': 'transparent',
'in': 'query',
'description': 'Background transparency of map (default=true).', # noqa
'required': False,
'schema': {
'type': 'boolean',
'default': True,
},
'style': 'form',
'explode': False
},
{'$ref': '#/components/parameters/bbox-crs-epsg'},
map_f
],
'responses': {
'200': {
'description': 'Response',
'content': {
'application/json': {}
}
},
'400': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/InvalidParameter"}, # noqa
'500': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/ServerError"}, # noqa
}
}
}
if mp.time_field is not None:
paths[pth]['get']['parameters'].append(
{'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/parameters/datetime"}) # noqa
return [{'name': 'maps'}], {'paths': paths}
pygeoapi/api/processes.py
+740
View File
@@ -0,0 +1,740 @@
# =================================================================
# Authors: Tom Kralidis <tomkralidis@gmail.com>
# Francesco Bartoli <xbartolone@gmail.com>
# Sander Schaminee <sander.schaminee@geocat.net>
# John A Stevenson <jostev@bgs.ac.uk>
# Colin Blackburn <colb@bgs.ac.uk>
# Ricardo Garcia Silva <ricardo.garcia.silva@geobeyond.it>
# Bernhard Mallinger <bernhard.mallinger@eox.at>
#
# Copyright (c) 2024 Tom Kralidis
# Copyright (c) 2022 Francesco Bartoli
# Copyright (c) 2022 John A Stevenson and Colin Blackburn
# Copyright (c) 2023 Ricardo Garcia Silva
# Copyright (c) 2024 Bernhard Mallinger
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without
# restriction, including without limitation the rights to use,
# copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following
# conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
#
# =================================================================
from copy import deepcopy
from datetime import datetime, timezone
from http import HTTPStatus
import json
import logging
from typing import Tuple
from pygeoapi import l10n
from pygeoapi.util import (
json_serial, render_j2_template, JobStatus, RequestedProcessExecutionMode,
to_json, DATETIME_FORMAT)
from pygeoapi.process.base import (
JobNotFoundError, JobResultNotFoundError, ProcessorExecuteError
)
from pygeoapi.process.manager.base import get_manager, Subscriber
from . import (
APIRequest, API, SYSTEM_LOCALE, F_JSON, FORMAT_TYPES, F_HTML, F_JSONLD,
)
LOGGER = logging.getLogger(__name__)
CONFORMANCE_CLASSES = [
'http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/ogc-process-description', # noqa
'http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/core',
'http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/json',
'http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/oas30',
'http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/callback'
]
def describe_processes(api: API, request: APIRequest,
process=None) -> Tuple[dict, int, str]:
"""
Provide processes metadata
:param request: A request object
:param process: process identifier, defaults to None to obtain
information about all processes
:returns: tuple of headers, status code, content
"""
processes = []
headers = request.get_response_headers(**api.api_headers)
if process is not None:
if process not in api.manager.processes.keys():
msg = 'Identifier not found'
return api.get_exception(
HTTPStatus.NOT_FOUND, headers,
request.format, 'NoSuchProcess', msg)
if len(api.manager.processes) > 0:
if process is not None:
relevant_processes = [process]
else:
LOGGER.debug('Processing limit parameter')
try:
limit = int(request.params.get('limit'))
if limit <= 0:
msg = 'limit value should be strictly positive'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
relevant_processes = list(api.manager.processes)[:limit]
except TypeError:
LOGGER.debug('returning all processes')
relevant_processes = api.manager.processes.keys()
except ValueError:
msg = 'limit value should be an integer'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
for key in relevant_processes:
p = api.manager.get_processor(key)
p2 = l10n.translate_struct(deepcopy(p.metadata),
request.locale)
p2['id'] = key
if process is None:
p2.pop('inputs')
p2.pop('outputs')
p2.pop('example', None)
p2['jobControlOptions'] = ['sync-execute']
if api.manager.is_async:
p2['jobControlOptions'].append('async-execute')
p2['outputTransmission'] = ['value']
p2['links'] = p2.get('links', [])
jobs_url = f"{api.base_url}/jobs"
process_url = f"{api.base_url}/processes/{key}"
# TODO translation support
link = {
'type': FORMAT_TYPES[F_JSON],
'rel': request.get_linkrel(F_JSON),
'href': f'{process_url}?f={F_JSON}',
'title': 'Process description as JSON',
'hreflang': api.default_locale
}
p2['links'].append(link)
link = {
'type': FORMAT_TYPES[F_HTML],
'rel': request.get_linkrel(F_HTML),
'href': f'{process_url}?f={F_HTML}',
'title': 'Process description as HTML',
'hreflang': api.default_locale
}
p2['links'].append(link)
link = {
'type': FORMAT_TYPES[F_HTML],
'rel': 'http://www.opengis.net/def/rel/ogc/1.0/job-list',
'href': f'{jobs_url}?f={F_HTML}',
'title': 'jobs for this process as HTML',
'hreflang': api.default_locale
}
p2['links'].append(link)
link = {
'type': FORMAT_TYPES[F_JSON],
'rel': 'http://www.opengis.net/def/rel/ogc/1.0/job-list',
'href': f'{jobs_url}?f={F_JSON}',
'title': 'jobs for this process as JSON',
'hreflang': api.default_locale
}
p2['links'].append(link)
link = {
'type': FORMAT_TYPES[F_JSON],
'rel': 'http://www.opengis.net/def/rel/ogc/1.0/execute',
'href': f'{process_url}/execution?f={F_JSON}',
'title': 'Execution for this process as JSON',
'hreflang': api.default_locale
}
p2['links'].append(link)
processes.append(p2)
if process is not None:
response = processes[0]
else:
process_url = f"{api.base_url}/processes"
response = {
'processes': processes,
'links': [{
'type': FORMAT_TYPES[F_JSON],
'rel': request.get_linkrel(F_JSON),
'title': 'This document as JSON',
'href': f'{process_url}?f={F_JSON}'
}, {
'type': FORMAT_TYPES[F_JSONLD],
'rel': request.get_linkrel(F_JSONLD),
'title': 'This document as RDF (JSON-LD)',
'href': f'{process_url}?f={F_JSONLD}'
}, {
'type': FORMAT_TYPES[F_HTML],
'rel': request.get_linkrel(F_HTML),
'title': 'This document as HTML',
'href': f'{process_url}?f={F_HTML}'
}]
}
if request.format == F_HTML: # render
if process is not None:
response = render_j2_template(api.tpl_config,
'processes/process.html',
response, request.locale)
else:
response = render_j2_template(api.tpl_config,
'processes/index.html', response,
request.locale)
return headers, HTTPStatus.OK, response
return headers, HTTPStatus.OK, to_json(response, api.pretty_print)
# TODO: get_jobs doesn't have tests
def get_jobs(api: API, request: APIRequest,
job_id=None) -> Tuple[dict, int, str]:
"""
Get process jobs
:param request: A request object
:param job_id: id of job
:returns: tuple of headers, status code, content
"""
headers = request.get_response_headers(SYSTEM_LOCALE,
**api.api_headers)
if job_id is None:
jobs = sorted(api.manager.get_jobs(),
key=lambda k: k['job_start_datetime'],
reverse=True)
else:
try:
jobs = [api.manager.get_job(job_id)]
except JobNotFoundError:
return api.get_exception(
HTTPStatus.NOT_FOUND, headers, request.format,
'InvalidParameterValue', job_id)
serialized_jobs = {
'jobs': [],
'links': [{
'href': f"{api.base_url}/jobs?f={F_HTML}",
'rel': request.get_linkrel(F_HTML),
'type': FORMAT_TYPES[F_HTML],
'title': 'Jobs list as HTML'
}, {
'href': f"{api.base_url}/jobs?f={F_JSON}",
'rel': request.get_linkrel(F_JSON),
'type': FORMAT_TYPES[F_JSON],
'title': 'Jobs list as JSON'
}]
}
for job_ in jobs:
job2 = {
'type': 'process',
'processID': job_['process_id'],
'jobID': job_['identifier'],
'status': job_['status'],
'message': job_['message'],
'progress': job_['progress'],
'parameters': job_.get('parameters'),
'job_start_datetime': job_['job_start_datetime'],
'job_end_datetime': job_['job_end_datetime']
}
# TODO: translate
if JobStatus[job_['status']] in (
JobStatus.successful, JobStatus.running, JobStatus.accepted):
job_result_url = f"{api.base_url}/jobs/{job_['identifier']}/results" # noqa
job2['links'] = [{
'href': f'{job_result_url}?f={F_HTML}',
'rel': 'http://www.opengis.net/def/rel/ogc/1.0/results',
'type': FORMAT_TYPES[F_HTML],
'title': f'results of job {job_id} as HTML'
}, {
'href': f'{job_result_url}?f={F_JSON}',
'rel': 'http://www.opengis.net/def/rel/ogc/1.0/results',
'type': FORMAT_TYPES[F_JSON],
'title': f'results of job {job_id} as JSON'
}]
if job_['mimetype'] not in (FORMAT_TYPES[F_JSON],
FORMAT_TYPES[F_HTML]):
job2['links'].append({
'href': job_result_url,
'rel': 'http://www.opengis.net/def/rel/ogc/1.0/results', # noqa
'type': job_['mimetype'],
'title': f"results of job {job_id} as {job_['mimetype']}" # noqa
})
serialized_jobs['jobs'].append(job2)
if job_id is None:
j2_template = 'jobs/index.html'
else:
serialized_jobs = serialized_jobs['jobs'][0]
j2_template = 'jobs/job.html'
if request.format == F_HTML:
data = {
'jobs': serialized_jobs,
'now': datetime.now(timezone.utc).strftime(DATETIME_FORMAT)
}
response = render_j2_template(api.tpl_config, j2_template, data,
request.locale)
return headers, HTTPStatus.OK, response
return headers, HTTPStatus.OK, to_json(serialized_jobs,
api.pretty_print)
def execute_process(api: API, request: APIRequest,
process_id) -> Tuple[dict, int, str]:
"""
Execute process
:param request: A request object
:param process_id: id of process
:returns: tuple of headers, status code, content
"""
# Responses are always in US English only
headers = request.get_response_headers(SYSTEM_LOCALE,
**api.api_headers)
if process_id not in api.manager.processes:
msg = 'identifier not found'
return api.get_exception(
HTTPStatus.NOT_FOUND, headers,
request.format, 'NoSuchProcess', msg)
data = request.data
if not data:
# TODO not all processes require input, e.g. time-dependent or
# random value generators
msg = 'missing request data'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'MissingParameterValue', msg)
try:
# Parse bytes data, if applicable
data = data.decode()
LOGGER.debug(data)
except (UnicodeDecodeError, AttributeError):
pass
try:
data = json.loads(data)
except (json.decoder.JSONDecodeError, TypeError) as err:
# Input does not appear to be valid JSON
LOGGER.error(err)
msg = 'invalid request data'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
data_dict = data.get('inputs', {})
LOGGER.debug(data_dict)
subscriber = None
subscriber_dict = data.get('subscriber')
if subscriber_dict:
try:
success_uri = subscriber_dict['successUri']
except KeyError:
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'MissingParameterValue', 'Missing successUri')
else:
subscriber = Subscriber(
# NOTE: successUri is mandatory according to the standard
success_uri=success_uri,
in_progress_uri=subscriber_dict.get('inProgressUri'),
failed_uri=subscriber_dict.get('failedUri'),
)
try:
execution_mode = RequestedProcessExecutionMode(
request.headers.get('Prefer', request.headers.get('prefer'))
)
except ValueError:
execution_mode = None
try:
LOGGER.debug('Executing process')
result = api.manager.execute_process(
process_id, data_dict, execution_mode=execution_mode,
subscriber=subscriber)
job_id, mime_type, outputs, status, additional_headers = result
headers.update(additional_headers or {})
headers['Location'] = f'{api.base_url}/jobs/{job_id}'
except ProcessorExecuteError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers,
request.format, err.ogc_exception_code, err.message)
response = {}
if status == JobStatus.failed:
response = outputs
if data.get('response', 'raw') == 'raw':
headers['Content-Type'] = mime_type
response = outputs
elif status not in (JobStatus.failed, JobStatus.accepted):
response['outputs'] = [outputs]
if status == JobStatus.accepted:
http_status = HTTPStatus.CREATED
elif status == JobStatus.failed:
http_status = HTTPStatus.BAD_REQUEST
else:
http_status = HTTPStatus.OK
if mime_type == 'application/json':
response2 = to_json(response, api.pretty_print)
else:
response2 = response
return headers, http_status, response2
def get_job_result(api: API, request: APIRequest,
job_id) -> Tuple[dict, int, str]:
"""
Get result of job (instance of a process)
:param request: A request object
:param job_id: ID of job
:returns: tuple of headers, status code, content
"""
headers = request.get_response_headers(SYSTEM_LOCALE,
**api.api_headers)
try:
job = api.manager.get_job(job_id)
except JobNotFoundError:
return api.get_exception(
HTTPStatus.NOT_FOUND, headers,
request.format, 'NoSuchJob', job_id
)
status = JobStatus[job['status']]
if status == JobStatus.running:
msg = 'job still running'
return api.get_exception(
HTTPStatus.NOT_FOUND, headers,
request.format, 'ResultNotReady', msg)
elif status == JobStatus.accepted:
# NOTE: this case is not mentioned in the specification
msg = 'job accepted but not yet running'
return api.get_exception(
HTTPStatus.NOT_FOUND, headers,
request.format, 'ResultNotReady', msg)
elif status == JobStatus.failed:
msg = 'job failed'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
try:
mimetype, job_output = api.manager.get_job_result(job_id)
except JobResultNotFoundError:
return api.get_exception(
HTTPStatus.INTERNAL_SERVER_ERROR, headers,
request.format, 'JobResultNotFound', job_id
)
if mimetype not in (None, FORMAT_TYPES[F_JSON]):
headers['Content-Type'] = mimetype
content = job_output
else:
if request.format == F_JSON:
content = json.dumps(job_output, sort_keys=True, indent=4,
default=json_serial)
else:
# HTML
headers['Content-Type'] = "text/html"
data = {
'job': {'id': job_id},
'result': job_output
}
content = render_j2_template(
api.config, 'jobs/results/index.html',
data, request.locale)
return headers, HTTPStatus.OK, content
def delete_job(
api: API, request: APIRequest, job_id
) -> Tuple[dict, int, str]:
"""
Delete a process job
:param job_id: job identifier
:returns: tuple of headers, status code, content
"""
response_headers = request.get_response_headers(
SYSTEM_LOCALE, **api.api_headers)
try:
success = api.manager.delete_job(job_id)
except JobNotFoundError:
return api.get_exception(
HTTPStatus.NOT_FOUND, response_headers, request.format,
'NoSuchJob', job_id
)
else:
if success:
http_status = HTTPStatus.OK
jobs_url = f"{api.base_url}/jobs"
response = {
'jobID': job_id,
'status': JobStatus.dismissed.value,
'message': 'Job dismissed',
'progress': 100,
'links': [{
'href': jobs_url,
'rel': 'up',
'type': FORMAT_TYPES[F_JSON],
'title': 'The job list for the current process'
}]
}
else:
return api.get_exception(
HTTPStatus.INTERNAL_SERVER_ERROR, response_headers,
request.format, 'InternalError', job_id
)
LOGGER.info(response)
# TODO: this response does not have any headers
return {}, http_status, response
def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str, dict]]: # noqa
"""
Get OpenAPI fragments
:param cfg: `dict` of configuration
:param locale: `str` of locale
:returns: `tuple` of `list` of tag objects, and `dict` of path objects
"""
from pygeoapi.openapi import OPENAPI_YAML
LOGGER.debug('setting up processes endpoints')
oas = {'tags': []}
paths = {}
process_manager = get_manager(cfg)
if len(process_manager.processes) > 0:
paths['/processes'] = {
'get': {
'summary': 'Processes',
'description': 'Processes',
'tags': ['server'],
'operationId': 'getProcesses',
'parameters': [
{'$ref': '#/components/parameters/f'}
],
'responses': {
'200': {'$ref': f"{OPENAPI_YAML['oapip']}/responses/ProcessList.yaml"}, # noqa
'default': {'$ref': '#/components/responses/default'}
}
}
}
LOGGER.debug('setting up processes')
for k, v in process_manager.processes.items():
if k.startswith('_'):
LOGGER.debug(f'Skipping hidden layer: {k}')
continue
name = l10n.translate(k, locale)
p = process_manager.get_processor(k)
md_desc = l10n.translate(p.metadata['description'], locale)
process_name_path = f'/processes/{name}'
tag = {
'name': name,
'description': md_desc,
'externalDocs': {}
}
for link in p.metadata.get('links', []):
if link['type'] == 'information':
translated_link = l10n.translate(link, locale)
tag['externalDocs']['description'] = translated_link[
'type']
tag['externalDocs']['url'] = translated_link['url']
break
if len(tag['externalDocs']) == 0:
del tag['externalDocs']
oas['tags'].append(tag)
paths[process_name_path] = {
'get': {
'summary': 'Get process metadata',
'description': md_desc,
'tags': [name],
'operationId': f'describe{name.capitalize()}Process',
'parameters': [
{'$ref': '#/components/parameters/f'}
],
'responses': {
'200': {'$ref': '#/components/responses/200'},
'default': {'$ref': '#/components/responses/default'}
}
}
}
paths[f'{process_name_path}/execution'] = {
'post': {
'summary': f"Process {l10n.translate(p.metadata['title'], locale)} execution", # noqa
'description': md_desc,
'tags': [name],
'operationId': f'execute{name.capitalize()}Job',
'responses': {
'200': {'$ref': '#/components/responses/200'},
'201': {'$ref': f"{OPENAPI_YAML['oapip']}/responses/ExecuteAsync.yaml"}, # noqa
'404': {'$ref': f"{OPENAPI_YAML['oapip']}/responses/NotFound.yaml"}, # noqa
'500': {'$ref': f"{OPENAPI_YAML['oapip']}/responses/ServerError.yaml"}, # noqa
'default': {'$ref': '#/components/responses/default'}
},
'requestBody': {
'description': 'Mandatory execute request JSON',
'required': True,
'content': {
'application/json': {
'schema': {
'$ref': f"{OPENAPI_YAML['oapip']}/schemas/execute.yaml" # noqa
}
}
}
}
}
}
if 'example' in p.metadata:
paths[f'{process_name_path}/execution']['post']['requestBody']['content']['application/json']['example'] = p.metadata['example'] # noqa
name_in_path = {
'name': 'jobId',
'in': 'path',
'description': 'job identifier',
'required': True,
'schema': {
'type': 'string'
}
}
paths['/jobs'] = {
'get': {
'summary': 'Retrieve jobs list',
'description': 'Retrieve a list of jobs',
'tags': ['jobs'],
'operationId': 'getJobs',
'responses': {
'200': {'$ref': '#/components/responses/200'},
'404': {'$ref': f"{OPENAPI_YAML['oapip']}/responses/NotFound.yaml"}, # noqa
'default': {'$ref': '#/components/responses/default'}
}
}
}
paths['/jobs/{jobId}'] = {
'get': {
'summary': 'Retrieve job details',
'description': 'Retrieve job details',
'tags': ['jobs'],
'parameters': [
name_in_path,
{'$ref': '#/components/parameters/f'}
],
'operationId': 'getJob',
'responses': {
'200': {'$ref': '#/components/responses/200'},
'404': {'$ref': f"{OPENAPI_YAML['oapip']}/responses/NotFound.yaml"}, # noqa
'default': {'$ref': '#/components/responses/default'}
}
},
'delete': {
'summary': 'Cancel / delete job',
'description': 'Cancel / delete job',
'tags': ['jobs'],
'parameters': [
name_in_path
],
'operationId': 'deleteJob',
'responses': {
'204': {'$ref': '#/components/responses/204'},
'404': {'$ref': f"{OPENAPI_YAML['oapip']}/responses/NotFound.yaml"}, # noqa
'default': {'$ref': '#/components/responses/default'}
}
},
}
paths['/jobs/{jobId}/results'] = {
'get': {
'summary': 'Retrieve job results',
'description': 'Retrive job resiults',
'tags': ['jobs'],
'parameters': [
name_in_path,
{'$ref': '#/components/parameters/f'}
],
'operationId': 'getJobResults',
'responses': {
'200': {'$ref': '#/components/responses/200'},
'404': {'$ref': f"{OPENAPI_YAML['oapip']}/responses/NotFound.yaml"}, # noqa
'default': {'$ref': '#/components/responses/default'}
}
}
}
return [{'name': 'proceses'}, {'name': 'jobs'}], {'paths': paths}
pygeoapi/api/stac.py
+256
View File
@@ -0,0 +1,256 @@
# =================================================================
# Authors: Tom Kralidis <tomkralidis@gmail.com>
# Francesco Bartoli <xbartolone@gmail.com>
# Sander Schaminee <sander.schaminee@geocat.net>
# John A Stevenson <jostev@bgs.ac.uk>
# Colin Blackburn <colb@bgs.ac.uk>
# Ricardo Garcia Silva <ricardo.garcia.silva@geobeyond.it>
# Bernhard Mallinger <bernhard.mallinger@eox.at>
#
# Copyright (c) 2024 Tom Kralidis
# Copyright (c) 2022 Francesco Bartoli
# Copyright (c) 2022 John A Stevenson and Colin Blackburn
# Copyright (c) 2023 Ricardo Garcia Silva
# Copyright (c) 2024 Bernhard Mallinger
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without
# restriction, including without limitation the rights to use,
# copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following
# conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
#
# =================================================================
from http import HTTPStatus
import logging
from typing import Tuple
from pygeoapi import l10n
from pygeoapi.plugin import load_plugin
from pygeoapi.provider.base import (
ProviderConnectionError, ProviderNotFoundError
)
from pygeoapi.util import (
get_provider_by_type, to_json, filter_dict_by_key_value,
render_j2_template
)
from . import APIRequest, API, FORMAT_TYPES, F_JSON, F_HTML
LOGGER = logging.getLogger(__name__)
CONFORMANCE_CLASSES = []
# TODO: no tests for this?
def get_stac_root(api: API, request: APIRequest) -> Tuple[dict, int, str]:
"""
Provide STAC root page
:param request: APIRequest instance with query params
:returns: tuple of headers, status code, content
"""
headers = request.get_response_headers(**api.api_headers)
id_ = 'pygeoapi-stac'
stac_version = '1.0.0-rc.2'
stac_url = f'{api.base_url}/stac'
content = {
'id': id_,
'type': 'Catalog',
'stac_version': stac_version,
'title': l10n.translate(
api.config['metadata']['identification']['title'],
request.locale),
'description': l10n.translate(
api.config['metadata']['identification']['description'],
request.locale),
'links': []
}
stac_collections = filter_dict_by_key_value(api.config['resources'],
'type', 'stac-collection')
for key, value in stac_collections.items():
content['links'].append({
'rel': 'child',
'href': f'{stac_url}/{key}?f={F_JSON}',
'type': FORMAT_TYPES[F_JSON]
})
content['links'].append({
'rel': 'child',
'href': f'{stac_url}/{key}',
'type': FORMAT_TYPES[F_HTML]
})
if request.format == F_HTML: # render
content = render_j2_template(api.tpl_config,
'stac/collection.html',
content, request.locale)
return headers, HTTPStatus.OK, content
return headers, HTTPStatus.OK, to_json(content, api.pretty_print)
# TODO: no tests for this?
def get_stac_path(api: API, request: APIRequest,
path) -> Tuple[dict, int, str]:
"""
Provide STAC resource path
:param request: APIRequest instance with query params
:returns: tuple of headers, status code, content
"""
headers = request.get_response_headers(**api.api_headers)
dataset = None
LOGGER.debug(f'Path: {path}')
dir_tokens = path.split('/')
if dir_tokens:
dataset = dir_tokens[0]
stac_collections = filter_dict_by_key_value(api.config['resources'],
'type', 'stac-collection')
if dataset not in stac_collections:
msg = 'Collection not found'
return api.get_exception(HTTPStatus.NOT_FOUND, headers,
request.format, 'NotFound', msg)
LOGGER.debug('Loading provider')
try:
p = load_plugin('provider', get_provider_by_type(
stac_collections[dataset]['providers'], 'stac'))
except ProviderConnectionError as err:
LOGGER.error(err)
msg = 'connection error (check logs)'
return api.get_exception(
HTTPStatus.INTERNAL_SERVER_ERROR, headers,
request.format, 'NoApplicableCode', msg)
id_ = f'{dataset}-stac'
stac_version = '1.0.0-rc.2'
content = {
'id': id_,
'type': 'Catalog',
'stac_version': stac_version,
'description': l10n.translate(
stac_collections[dataset]['description'], request.locale),
'links': []
}
try:
stac_data = p.get_data_path(
f'{api.base_url}/stac',
path,
path.replace(dataset, '', 1)
)
except ProviderNotFoundError as err:
LOGGER.error(err)
msg = 'resource not found'
return api.get_exception(HTTPStatus.NOT_FOUND, headers,
request.format, 'NotFound', msg)
except Exception as err:
LOGGER.error(err)
msg = 'data query error'
return api.get_exception(
HTTPStatus.INTERNAL_SERVER_ERROR, headers,
request.format, 'NoApplicableCode', msg)
if isinstance(stac_data, dict):
content.update(stac_data)
content['links'].extend(
stac_collections[dataset].get('links', []))
if request.format == F_HTML: # render
content['path'] = path
if 'assets' in content: # item view
if content['type'] == 'Collection':
content = render_j2_template(
api.tpl_config,
'stac/collection_base.html',
content,
request.locale
)
elif content['type'] == 'Feature':
content = render_j2_template(
api.tpl_config,
'stac/item.html',
content,
request.locale
)
else:
msg = f'Unknown STAC type {content.type}'
LOGGER.error(msg)
return api.get_exception(
HTTPStatus.INTERNAL_SERVER_ERROR,
headers,
request.format,
'NoApplicableCode',
msg)
else:
content = render_j2_template(api.tpl_config,
'stac/catalog.html',
content, request.locale)
return headers, HTTPStatus.OK, content
return headers, HTTPStatus.OK, to_json(content, api.pretty_print)
else: # send back file
headers.pop('Content-Type', None)
return headers, HTTPStatus.OK, stac_data
def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str, dict]]: # noqa
"""
Get OpenAPI fragments
:param cfg: `dict` of configuration
:param locale: `str` of locale
:returns: `tuple` of `list` of tag objects, and `dict` of path objects
"""
LOGGER.debug('setting up STAC')
stac_collections = filter_dict_by_key_value(cfg['resources'],
'type', 'stac-collection')
paths = {}
if stac_collections:
paths['/stac'] = {
'get': {
'summary': 'SpatioTemporal Asset Catalog',
'description': 'SpatioTemporal Asset Catalog',
'tags': ['stac'],
'operationId': 'getStacCatalog',
'parameters': [],
'responses': {
'200': {'$ref': '#/components/responses/200'},
'default': {'$ref': '#/components/responses/default'}
}
}
}
return [{'name': 'stac'}], {'paths': paths}
pygeoapi/api/tiles.py
+534
View File
@@ -0,0 +1,534 @@
# =================================================================
# Authors: Tom Kralidis <tomkralidis@gmail.com>
# Francesco Bartoli <xbartolone@gmail.com>
# Sander Schaminee <sander.schaminee@geocat.net>
# John A Stevenson <jostev@bgs.ac.uk>
# Colin Blackburn <colb@bgs.ac.uk>
# Ricardo Garcia Silva <ricardo.garcia.silva@geobeyond.it>
# Bernhard Mallinger <bernhard.mallinger@eox.at>
#
# Copyright (c) 2024 Tom Kralidis
# Copyright (c) 2022 Francesco Bartoli
# Copyright (c) 2022 John A Stevenson and Colin Blackburn
# Copyright (c) 2023 Ricardo Garcia Silva
# Copyright (c) 2024 Bernhard Mallinger
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without
# restriction, including without limitation the rights to use,
# copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following
# conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
#
# =================================================================
import logging
from http import HTTPStatus
from typing import Tuple
from pygeoapi import l10n
from pygeoapi.plugin import load_plugin
from pygeoapi.models.provider.base import (TilesMetadataFormat,
TileMatrixSetEnum)
from pygeoapi.provider.base import (
ProviderGenericError, ProviderTypeError
)
from pygeoapi.util import (
get_provider_by_type, to_json, filter_dict_by_key_value,
filter_providers_by_type, render_j2_template
)
from . import (
APIRequest, API, FORMAT_TYPES, F_JSON, F_HTML, SYSTEM_LOCALE, F_JSONLD
)
LOGGER = logging.getLogger(__name__)
CONFORMANCE_CLASSES = [
'http://www.opengis.net/spec/ogcapi-tiles-1/1.0/conf/core',
'http://www.opengis.net/spec/ogcapi-tiles-1/1.0/conf/mvt',
'http://www.opengis.net/spec/ogcapi-tiles-1/1.0/conf/tileset',
'http://www.opengis.net/spec/ogcapi-tiles-1/1.0/conf/tilesets-list',
'http://www.opengis.net/spec/ogcapi-tiles-1/1.0/conf/oas30',
'http://www.opengis.net/spec/ogcapi-tiles-1/1.0/conf/geodata-tilesets'
]
def get_collection_tiles(api: API, request: APIRequest,
dataset=None) -> Tuple[dict, int, str]:
"""
Provide collection tiles
:param request: A request object
:param dataset: name of collection
:returns: tuple of headers, status code, content
"""
headers = request.get_response_headers(SYSTEM_LOCALE,
**api.api_headers)
if any([dataset is None,
dataset not in api.config['resources'].keys()]):
msg = 'Collection not found'
return api.get_exception(
HTTPStatus.NOT_FOUND, headers, request.format, 'NotFound', msg)
LOGGER.debug('Creating collection tiles')
LOGGER.debug('Loading provider')
try:
t = get_provider_by_type(
api.config['resources'][dataset]['providers'], 'tile')
p = load_plugin('provider', t)
except (KeyError, ProviderTypeError):
msg = 'Invalid collection tiles'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
except ProviderGenericError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers, request.format,
err.ogc_exception_code, err.message)
tiles = {
'links': [],
'tilesets': []
}
tiles['links'].append({
'type': FORMAT_TYPES[F_JSON],
'rel': request.get_linkrel(F_JSON),
'title': 'This document as JSON',
'href': f'{api.get_collections_url()}/{dataset}/tiles?f={F_JSON}'
})
tiles['links'].append({
'type': FORMAT_TYPES[F_JSONLD],
'rel': request.get_linkrel(F_JSONLD),
'title': 'This document as RDF (JSON-LD)',
'href': f'{api.get_collections_url()}/{dataset}/tiles?f={F_JSONLD}'
})
tiles['links'].append({
'type': FORMAT_TYPES[F_HTML],
'rel': request.get_linkrel(F_HTML),
'title': 'This document as HTML',
'href': f'{api.get_collections_url()}/{dataset}/tiles?f={F_HTML}'
})
tile_services = p.get_tiles_service(
baseurl=api.base_url,
servicepath=f'{api.get_collections_url()}/{dataset}/tiles/{{tileMatrixSetId}}/{{tileMatrix}}/{{tileRow}}/{{tileCol}}?f={p.format_type}' # noqa
)
for service in tile_services['links']:
tiles['links'].append(service)
tiling_schemes = p.get_tiling_schemes()
for matrix in tiling_schemes:
tile_matrix = {
'title': dataset,
'tileMatrixSetURI': matrix.tileMatrixSetURI,
'crs': matrix.crs,
'dataType': 'vector',
'links': []
}
tile_matrix['links'].append({
'type': FORMAT_TYPES[F_JSON],
'rel': 'http://www.opengis.net/def/rel/ogc/1.0/tiling-scheme',
'title': f'{matrix.tileMatrixSet} TileMatrixSet definition (as {F_JSON})', # noqa
'href': f'{api.base_url}/TileMatrixSets/{matrix.tileMatrixSet}?f={F_JSON}' # noqa
})
tile_matrix['links'].append({
'type': FORMAT_TYPES[F_JSON],
'rel': request.get_linkrel(F_JSON),
'title': f'{dataset} - {matrix.tileMatrixSet} - {F_JSON}',
'href': f'{api.get_collections_url()}/{dataset}/tiles/{matrix.tileMatrixSet}?f={F_JSON}' # noqa
})
tile_matrix['links'].append({
'type': FORMAT_TYPES[F_HTML],
'rel': request.get_linkrel(F_HTML),
'title': f'{dataset} - {matrix.tileMatrixSet} - {F_HTML}',
'href': f'{api.get_collections_url()}/{dataset}/tiles/{matrix.tileMatrixSet}?f={F_HTML}' # noqa
})
tiles['tilesets'].append(tile_matrix)
if request.format == F_HTML: # render
tiles['id'] = dataset
tiles['title'] = l10n.translate(
api.config['resources'][dataset]['title'], SYSTEM_LOCALE)
tiles['tilesets'] = [
scheme.tileMatrixSet for scheme in p.get_tiling_schemes()]
tiles['bounds'] = \
api.config['resources'][dataset]['extents']['spatial']['bbox']
tiles['minzoom'] = p.options['zoom']['min']
tiles['maxzoom'] = p.options['zoom']['max']
tiles['collections_path'] = api.get_collections_url()
tiles['tile_type'] = p.tile_type
content = render_j2_template(api.tpl_config,
'collections/tiles/index.html', tiles,
request.locale)
return headers, HTTPStatus.OK, content
return headers, HTTPStatus.OK, to_json(tiles, api.pretty_print)
# TODO: no test for this function?
def get_collection_tiles_data(
api: API, request: APIRequest,
dataset=None, matrix_id=None,
z_idx=None, y_idx=None, x_idx=None) -> Tuple[dict, int, str]:
"""
Get collection items tiles
:param request: A request object
:param dataset: dataset name
:param matrix_id: matrix identifier
:param z_idx: z index
:param y_idx: y index
:param x_idx: x index
:returns: tuple of headers, status code, content
"""
format_ = request.format
if not format_:
return api.get_format_exception(request)
headers = request.get_response_headers(SYSTEM_LOCALE,
**api.api_headers)
LOGGER.debug('Processing tiles')
collections = filter_dict_by_key_value(api.config['resources'],
'type', 'collection')
if dataset not in collections.keys():
msg = 'Collection not found'
return api.get_exception(
HTTPStatus.NOT_FOUND, headers, request.format, 'NotFound', msg)
LOGGER.debug('Loading tile provider')
try:
t = get_provider_by_type(
api.config['resources'][dataset]['providers'], 'tile')
p = load_plugin('provider', t)
format_ = p.format_type
headers['Content-Type'] = format_
LOGGER.debug(f'Fetching tileset id {matrix_id} and tile {z_idx}/{y_idx}/{x_idx}') # noqa
content = p.get_tiles(layer=p.get_layer(), tileset=matrix_id,
z=z_idx, y=y_idx, x=x_idx, format_=format_)
if content is None:
msg = 'identifier not found'
return api.get_exception(
HTTPStatus.NOT_FOUND, headers, format_, 'NotFound', msg)
else:
return headers, HTTPStatus.OK, content
# @TODO: figure out if the spec requires to return json errors
except KeyError:
msg = 'Invalid collection tiles'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, format_,
'InvalidParameterValue', msg)
except ProviderGenericError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers, request.format,
err.ogc_exception_code, err.message)
# TODO: no test for this function?
def get_collection_tiles_metadata(
api: API, request: APIRequest,
dataset=None, matrix_id=None) -> Tuple[dict, int, str]:
"""
Get collection items tiles
:param request: A request object
:param dataset: dataset name
:param matrix_id: matrix identifier
:returns: tuple of headers, status code, content
"""
if not request.is_valid([TilesMetadataFormat.TILEJSON]):
return api.get_format_exception(request)
headers = request.get_response_headers(**api.api_headers)
if any([dataset is None,
dataset not in api.config['resources'].keys()]):
msg = 'Collection not found'
return api.get_exception(
HTTPStatus.NOT_FOUND, headers, request.format, 'NotFound', msg)
LOGGER.debug('Creating collection tiles')
LOGGER.debug('Loading provider')
try:
t = get_provider_by_type(
api.config['resources'][dataset]['providers'], 'tile')
p = load_plugin('provider', t)
except KeyError:
msg = 'Invalid collection tiles'
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', msg)
except ProviderGenericError as err:
LOGGER.error(err)
return api.get_exception(
err.http_status_code, headers, request.format,
err.ogc_exception_code, err.message)
# Get provider language (if any)
prv_locale = l10n.get_plugin_locale(t, request.raw_locale)
# Set response language to requested provider locale
# (if it supports language) and/or otherwise the requested pygeoapi
# locale (or fallback default locale)
l10n.set_response_language(headers, prv_locale, request.locale)
tiles_metadata = p.get_metadata(
dataset=dataset, server_url=api.base_url,
layer=p.get_layer(), tileset=matrix_id,
metadata_format=request._format, title=l10n.translate(
api.config['resources'][dataset]['title'],
request.locale),
description=l10n.translate(
api.config['resources'][dataset]['description'],
request.locale),
language=prv_locale)
if request.format == F_HTML: # render
content = render_j2_template(api.tpl_config,
'collections/tiles/metadata.html',
tiles_metadata, request.locale)
return headers, HTTPStatus.OK, content
else:
return headers, HTTPStatus.OK, tiles_metadata
def tilematrixsets(api: API,
request: APIRequest) -> Tuple[dict, int, str]:
"""
Provide tileMatrixSets definition
:param request: A request object
:returns: tuple of headers, status code, content
"""
headers = request.get_response_headers(**api.api_headers)
# Retrieve available TileMatrixSets
enums = [e.value for e in TileMatrixSetEnum]
tms = {"tileMatrixSets": []}
for e in enums:
tms['tileMatrixSets'].append({
"title": e.title,
"id": e.tileMatrixSet,
"uri": e.tileMatrixSetURI,
"links": [
{
"rel": "self",
"type": "text/html",
"title": f"The HTML representation of the {e.tileMatrixSet} tile matrix set", # noqa
"href": f"{api.base_url}/TileMatrixSets/{e.tileMatrixSet}?f=html" # noqa
},
{
"rel": "self",
"type": "application/json",
"title": f"The JSON representation of the {e.tileMatrixSet} tile matrix set", # noqa
"href": f"{api.base_url}/TileMatrixSets/{e.tileMatrixSet}?f=json" # noqa
}
]
})
tms['links'] = [{
"rel": "alternate",
"type": "text/html",
"title": "This document as HTML",
"href": f"{api.base_url}/tileMatrixSets?f=html"
}, {
"rel": "self",
"type": "application/json",
"title": "This document",
"href": f"{api.base_url}/tileMatrixSets?f=json"
}]
if request.format == F_HTML: # render
content = render_j2_template(api.tpl_config,
'tilematrixsets/index.html',
tms, request.locale)
return headers, HTTPStatus.OK, content
return headers, HTTPStatus.OK, to_json(tms, api.pretty_print)
def tilematrixset(api: API,
request: APIRequest,
tileMatrixSetId) -> Tuple[dict,
int, str]:
"""
Provide tile matrix definition
:param request: A request object
:returns: tuple of headers, status code, content
"""
headers = request.get_response_headers(**api.api_headers)
# Retrieve relevant TileMatrixSet
enums = [e.value for e in TileMatrixSetEnum]
enum = None
try:
for e in enums:
if tileMatrixSetId == e.tileMatrixSet:
enum = e
if not enum:
raise ValueError('could not find this tilematrixset')
except ValueError as err:
return api.get_exception(
HTTPStatus.BAD_REQUEST, headers, request.format,
'InvalidParameterValue', str(err))
tms = {
"title": enum.tileMatrixSet,
"crs": enum.crs,
"id": enum.tileMatrixSet,
"uri": enum.tileMatrixSetURI,
"orderedAxes": enum.orderedAxes,
"wellKnownScaleSet": enum.wellKnownScaleSet,
"tileMatrices": enum.tileMatrices
}
if request.format == F_HTML: # render
content = render_j2_template(api.tpl_config,
'tilematrixsets/tilematrixset.html',
tms, request.locale)
return headers, HTTPStatus.OK, content
return headers, HTTPStatus.OK, to_json(tms, api.pretty_print)
def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str, dict]]: # noqa
"""
Get OpenAPI fragments
:param cfg: `dict` of configuration
:param locale: `str` of locale
:returns: `tuple` of `list` of tag objects, and `dict` of path objects
"""
from pygeoapi.openapi import OPENAPI_YAML, get_visible_collections
paths = {}
LOGGER.debug('setting up tiles endpoints')
collections = filter_dict_by_key_value(cfg['resources'],
'type', 'collection')
for k, v in get_visible_collections(cfg).items():
tile_extension = filter_providers_by_type(
collections[k]['providers'], 'tile')
if tile_extension:
tp = load_plugin('provider', tile_extension)
tiles_path = f'/collections/{k}/tiles'
title = l10n.translate(v['title'], locale)
description = l10n.translate(v['description'], locale)
paths[tiles_path] = {
'get': {
'summary': f'Fetch a {title} tiles description',
'description': description,
'tags': [k],
'operationId': f'describe{k.capitalize()}Tiles',
'parameters': [
{'$ref': '#/components/parameters/f'},
{'$ref': '#/components/parameters/lang'}
],
'responses': {
'200': {'$ref': '#/components/responses/Tiles'},
'400': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/InvalidParameter"}, # noqa
'404': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/NotFound"}, # noqa
'500': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/ServerError"} # noqa
}
}
}
tiles_data_path = f'{tiles_path}/{{tileMatrixSetId}}/{{tileMatrix}}/{{tileRow}}/{{tileCol}}' # noqa
paths[tiles_data_path] = {
'get': {
'summary': f'Get a {title} tile',
'description': description,
'tags': [k],
'operationId': f'get{k.capitalize()}Tiles',
'parameters': [
{'$ref': f"{OPENAPI_YAML['oapit']}#/components/parameters/tileMatrixSetId"}, # noqa
{'$ref': f"{OPENAPI_YAML['oapit']}#/components/parameters/tileMatrix"}, # noqa
{'$ref': f"{OPENAPI_YAML['oapit']}#/components/parameters/tileRow"}, # noqa
{'$ref': f"{OPENAPI_YAML['oapit']}#/components/parameters/tileCol"}, # noqa
{
'name': 'f',
'in': 'query',
'description': 'The optional f parameter indicates the output format which the server shall provide as part of the response document.', # noqa
'required': False,
'schema': {
'type': 'string',
'enum': [tp.format_type],
'default': tp.format_type
},
'style': 'form',
'explode': False
}
],
'responses': {
'400': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/InvalidParameter"}, # noqa
'404': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/NotFound"}, # noqa
'500': {'$ref': f"{OPENAPI_YAML['oapif-1']}#/components/responses/ServerError"} # noqa
}
}
}
mimetype = tile_extension['format']['mimetype']
paths[tiles_data_path]['get']['responses']['200'] = {
'description': 'successful operation',
'content': {
mimetype: {
'schema': {
'type': 'string',
'format': 'binary'
}
}
}
}
return [{'name': 'tiles'}], {'paths': paths}