diff --git a/pygeoapi/config.py b/pygeoapi/config.py index 88f50e74b..3f5ade823 100644 --- a/pygeoapi/config.py +++ b/pygeoapi/config.py @@ -41,25 +41,30 @@ LOGGER = logging.getLogger(__name__) -def get_config(raw: bool = False) -> dict: +def get_config(config_path: str = None, raw: bool = False) -> dict: """ Get pygeoapi configurations + :param config_path: `str` of configuration file location, if `None` + will attempt to read from `PYGEOAPI_CONFIG` :param raw: `bool` over interpolation during config loading :returns: `dict` of pygeoapi configuration """ - if not os.environ.get('PYGEOAPI_CONFIG'): - raise RuntimeError('PYGEOAPI_CONFIG environment variable not set') + if config_path is None: + config_path = os.environ.get('PYGEOAPI_CONFIG') - with open(os.environ.get('PYGEOAPI_CONFIG'), encoding='utf8') as fh: + if config_path is None: + raise KeyError('PYGEOAPI_CONFIG environment variable not set') + + with open(config_path, encoding='utf8') as fh: if raw: - CONFIG = yaml.safe_load(fh) + config = yaml.safe_load(fh) else: - CONFIG = yaml_load(fh) + config = yaml_load(fh) - return CONFIG + return config def load_schema() -> dict: diff --git a/pygeoapi/flask_app.py b/pygeoapi/flask_app.py index f659655fc..9e3673e15 100644 --- a/pygeoapi/flask_app.py +++ b/pygeoapi/flask_app.py @@ -31,7 +31,7 @@ """Flask module providing the route paths to the api""" import os -from typing import Callable, Union +from typing import Union import click from flask import (Flask, Blueprint, make_response, request, @@ -51,613 +51,677 @@ from pygeoapi.util import get_mimetype, get_api_rules -CONFIG = get_config() -OPENAPI = load_openapi_document() - -API_RULES = get_api_rules(CONFIG) - -if CONFIG['server'].get('admin'): - import pygeoapi.api.admin as admin_api - from pygeoapi.api.admin import Admin - -STATIC_FOLDER = 'static' -if 'templates' in CONFIG['server']: - STATIC_FOLDER = CONFIG['server']['templates'].get('static', 'static') - -APP = Flask(__name__, static_folder=STATIC_FOLDER, static_url_path='/static') -APP.url_map.strict_slashes = API_RULES.strict_slashes - -BLUEPRINT = Blueprint( - 'pygeoapi', - __name__, - static_folder=STATIC_FOLDER, - url_prefix=API_RULES.get_url_prefix('flask') -) -ADMIN_BLUEPRINT = Blueprint( - 'admin', - __name__, - static_folder=STATIC_FOLDER, - url_prefix=API_RULES.get_url_prefix('flask') -) - -# CORS: optionally enable from config. -if CONFIG['server'].get('cors', False): - try: - from flask_cors import CORS - CORS(APP, CORS_EXPOSE_HEADERS=['*']) - except ModuleNotFoundError: - print('Python package flask-cors required for CORS support') - -APP.config['JSONIFY_PRETTYPRINT_REGULAR'] = CONFIG['server'].get( - 'pretty_print', True) - -api_ = API(CONFIG, OPENAPI) - -OGC_SCHEMAS_LOCATION = CONFIG['server'].get('ogc_schemas_location') - -if (OGC_SCHEMAS_LOCATION is not None and - not OGC_SCHEMAS_LOCATION.startswith('http')): - # serve the OGC schemas locally +# Function to return a WSGI application, +# passing the locations for the config and +# openapi files as variables, instead as environment variables +def make_wsgi_app(config_location: str, openapi_location: str) -> Flask: + """ + Create a WSGI application + Args: + config_location (str): location of the pygeoapi config file + openapi_location (str): location of the OpenAPI document file - if not os.path.exists(OGC_SCHEMAS_LOCATION): - raise RuntimeError('OGC schemas misconfigured') + Returns: + Flask WSGI application + """ + config = get_config(config_path=config_location) + openapi = load_openapi_document(pygeoapi_openapi=openapi_location) - @BLUEPRINT.route('/schemas/', methods=['GET']) - def schemas(path: str): - """ - Serve OGC schemas locally + api_rules = get_api_rules(config) - :param path: path of the OGC schema document + if config['server'].get('admin'): + import pygeoapi.api.admin as admin_api + from pygeoapi.api.admin import Admin - :returns: HTTP response - """ + static_folder = 'static' + if 'templates' in config['server']: + static_folder = config['server']['templates'].get('static', 'static') - full_filepath = os.path.join(OGC_SCHEMAS_LOCATION, path) - dirname_ = os.path.dirname(full_filepath) - basename_ = os.path.basename(full_filepath) + app = Flask( + __name__, + static_folder=static_folder, + static_url_path='/static' + ) - path_ = dirname_.replace('..', '').replace('//', '').replace('./', '') + app.url_map.strict_slashes = api_rules.strict_slashes - if '..' in path_: - return 'Invalid path', 400 + blueprint = Blueprint( + 'pygeoapi', + __name__, + static_folder=static_folder, + url_prefix=api_rules.get_url_prefix('flask') + ) + admin_blueprint = Blueprint( + 'admin', + __name__, + static_folder=static_folder, + url_prefix=api_rules.get_url_prefix('flask') + ) - return send_from_directory(path_, basename_, - mimetype=get_mimetype(basename_)) + # CORS: optionally enable from config. + if config['server'].get('cors', False): + try: + from flask_cors import CORS + CORS(app, CORS_EXPOSE_HEADERS=['*']) + except ModuleNotFoundError: + print('Python package flask-cors required for CORS support') + app.config['JSONIFY_PRETTYPRINT_REGULAR'] = config['server'].get( + 'pretty_print', True) -def execute_from_flask(api_function: Callable, request: Request, *args, - skip_valid_check: bool = False, - alternative_api: API | None = None - ) -> Response: - """ - Executes API function from Flask + api_ = API(config, openapi) - :param api_function: API function - :param request: request object - :param *args: variable length additional arguments - :param skip_validity_check: bool - :param alternative_api: specify custom api instance such as Admin + ogc_schemas_location = config['server'].get('ogc_schemas_location') - :returns: A Response instance - """ + if (ogc_schemas_location is not None and + not ogc_schemas_location.startswith('http')): + # serve the OGC schemas locally - actual_api = api_ if alternative_api is None else alternative_api + if not os.path.exists(ogc_schemas_location): + raise RuntimeError('OGC schemas misconfigured') - api_request = APIRequest.from_flask(request, actual_api.locales) + @blueprint.route('/schemas/', methods=['GET']) + def schemas(path: str) -> Response: + """ + Serve OGC schemas locally - content: Union[str, bytes] + :param path: path of the OGC schema document - if not skip_valid_check and not api_request.is_valid(): - headers, status, content = actual_api.get_format_exception(api_request) - else: - headers, status, content = api_function(actual_api, api_request, *args) - content = apply_gzip(headers, content) + :returns: HTTP response + """ - response = make_response(content, status) + full_filepath = os.path.join(ogc_schemas_location, path) + dirname_ = os.path.dirname(full_filepath) + basename_ = os.path.basename(full_filepath) - if headers: - response.headers = headers - return response + path_ = dirname_.replace('..', '').replace('//', '').replace('./', '') # noqa: E501 + if '..' in path_: + return 'Invalid path', 400 -@BLUEPRINT.route('/') -def landing_page(): - """ - OGC API landing page endpoint + return send_from_directory( + path_, + basename_, + mimetype=get_mimetype(basename_) + ) - :returns: HTTP response - """ - return execute_from_flask(core_api.landing_page, request) + def execute_from_flask( + api_function: callable, + request: Request, + *args, + skip_valid_check: bool = False, + alternative_api: Response = None, + ) -> Response: + """ + Executes API function from Flask + :param api_function: API function + :param request: request object + :param *args: variable length additional arguments + :param skip_validity_check: bool + :param alternative_api: specify custom api instance such as Admin -@BLUEPRINT.route('/openapi') -def openapi(): - """ - OpenAPI endpoint + :returns: A Response instance + """ - :returns: HTTP response - """ + actual_api = api_ if alternative_api is None else alternative_api - return execute_from_flask(core_api.openapi_, request) + api_request = APIRequest.from_flask(request, actual_api.locales) + content: Union[str, bytes] -@BLUEPRINT.route('/conformance') -def conformance(): - """ - OGC API conformance endpoint + if not skip_valid_check and not api_request.is_valid(): + headers, status, content = \ + actual_api.get_format_exception(api_request) + else: + headers, status, content = \ + api_function(actual_api, api_request, *args) + content = apply_gzip(headers, content) - :returns: HTTP response - """ + response = make_response(content, status) - return execute_from_flask(core_api.conformance, request) + if headers: + response.headers = headers + return response + @blueprint.route('/') + def landing_page() -> Response: + """ + OGC API landing page endpoint -@BLUEPRINT.route('/TileMatrixSets/') -def get_tilematrix_set(tileMatrixSetId: str | None = None): - """ - OGC API TileMatrixSet endpoint - - :param tileMatrixSetId: identifier of tile matrix set + :returns: HTTP response + """ + return execute_from_flask(core_api.landing_page, request) - :returns: HTTP response - """ + @blueprint.route('/openapi') + def openapi() -> Response: + """ + OpenAPI endpoint - return execute_from_flask(tiles_api.tilematrixset, request, - tileMatrixSetId) + :returns: HTTP response + """ + return execute_from_flask(core_api.openapi_, request) -@BLUEPRINT.route('/TileMatrixSets') -def get_tilematrix_sets(): - """ - OGC API TileMatrixSets endpoint + @blueprint.route('/conformance') + def conformance() -> Response: + """ + OGC API conformance endpoint - :returns: HTTP response - """ + :returns: HTTP response + """ - return execute_from_flask(tiles_api.tilematrixsets, request) + return execute_from_flask(core_api.conformance, request) + @blueprint.route('/TileMatrixSets/') + def get_tilematrix_set(tileMatrixSetId: str) -> Response: + """ + OGC API TileMatrixSet endpoint -@BLUEPRINT.route('/collections') -@BLUEPRINT.route('/collections/') -def collections(collection_id: str | None = None): - """ - OGC API collections endpoint + :param tileMatrixSetId: identifier of tile matrix set - :param collection_id: collection identifier + :returns: HTTP response + """ + return execute_from_flask( + tiles_api.tilematrixset, request, tileMatrixSetId + ) - :returns: HTTP response - """ + @blueprint.route('/TileMatrixSets') + def get_tilematrix_sets() -> Response: + """ + OGC API TileMatrixSets endpoint - return execute_from_flask(core_api.describe_collections, request, - collection_id) + :returns: HTTP response + """ + return execute_from_flask(tiles_api.tilematrixsets, request) + @blueprint.route('/collections') + @blueprint.route('/collections/') + def collections(collection_id: str = None) -> Response: + """ + OGC API collections endpoint -@BLUEPRINT.route('/collections//schema') -def collection_schema(collection_id: str | None = None): - """ - OGC API - collections schema endpoint + :param collection_id: collection identifier - :param collection_id: collection identifier + :returns: HTTP response + """ + return execute_from_flask( + core_api.describe_collections, request, collection_id + ) - :returns: HTTP response - """ + @blueprint.route('/collections//schema') + def collection_schema(collection_id: str) -> Response: + """ + OGC API - collections schema endpoint - return execute_from_flask(core_api.get_collection_schema, request, - collection_id) + :param collection_id: collection identifier + :returns: HTTP response + """ -@BLUEPRINT.route('/collections//queryables') -def collection_queryables(collection_id: str | None = None): - """ - OGC API collections queryables endpoint + return execute_from_flask( + core_api.get_collection_schema, request, collection_id + ) - :param collection_id: collection identifier + @blueprint.route('/collections//queryables') + def collection_queryables(collection_id: str) -> Response: + """ + OGC API collections queryables endpoint - :returns: HTTP response - """ + :param collection_id: collection identifier - return execute_from_flask(itemtypes_api.get_collection_queryables, request, - collection_id) + :returns: HTTP response + """ + return execute_from_flask( + itemtypes_api.get_collection_queryables, request, collection_id + ) -@BLUEPRINT.route('/collections//items', - methods=['GET', 'POST', 'OPTIONS'], - provide_automatic_options=False) -@BLUEPRINT.route('/collections//items/', - methods=['GET', 'PUT', 'DELETE', 'OPTIONS'], - provide_automatic_options=False) -def collection_items(collection_id: str, item_id: str | None = None): - """ - OGC API collections items endpoint + @blueprint.route( + '/collections//items', + methods=['GET', 'POST', 'OPTIONS'], + provide_automatic_options=False + ) + def collection_items(collection_id: str) -> Response: + """ + OGC API collections items endpoint - :param collection_id: collection identifier - :param item_id: item identifier + :param collection_id: collection identifier - :returns: HTTP response - """ + :returns: HTTP response + """ - if item_id is None: if request.method == 'POST': # filter or manage items - if request.content_type is not None: - if request.content_type == 'application/geo+json': - return execute_from_flask( - itemtypes_api.manage_collection_item, - request, 'create', collection_id, - skip_valid_check=True) - else: - return execute_from_flask( - itemtypes_api.get_collection_items, request, - collection_id, skip_valid_check=True) + if request.content_type == 'application/geo+json': + return execute_from_flask( + itemtypes_api.manage_collection_item, + request, 'create', collection_id, + skip_valid_check=True + ) + else: + return execute_from_flask( + itemtypes_api.get_collection_items, request, + collection_id, skip_valid_check=True + ) + elif request.method == 'OPTIONS': return execute_from_flask( - itemtypes_api.manage_collection_item, request, 'options', - collection_id, skip_valid_check=True) - else: # GET: list items - return execute_from_flask(itemtypes_api.get_collection_items, - request, collection_id, - skip_valid_check=True) - - elif request.method == 'DELETE': - return execute_from_flask(itemtypes_api.manage_collection_item, - request, 'delete', collection_id, item_id, - skip_valid_check=True) - elif request.method == 'PUT': - return execute_from_flask(itemtypes_api.manage_collection_item, - request, 'update', collection_id, item_id, - skip_valid_check=True) - elif request.method == 'OPTIONS': - return execute_from_flask(itemtypes_api.manage_collection_item, - request, 'options', collection_id, item_id, - skip_valid_check=True) - else: - return execute_from_flask(itemtypes_api.get_collection_item, request, - collection_id, item_id) - - -@BLUEPRINT.route('/collections//coverage') -def collection_coverage(collection_id: str): - """ - OGC API - Coverages coverage endpoint - - :param collection_id: collection identifier - - :returns: HTTP response - """ - - return execute_from_flask(coverages_api.get_collection_coverage, request, - collection_id, skip_valid_check=True) + itemtypes_api.manage_collection_item, request, 'options', + collection_id, skip_valid_check=True + ) + # GET: list items + return execute_from_flask( + itemtypes_api.get_collection_items, + request, collection_id, + skip_valid_check=True + ) -@BLUEPRINT.route('/collections//tiles') -def get_collection_tiles(collection_id: str | None = None): - """ - OGC open api collections tiles access point - - :param collection_id: collection identifier - - :returns: HTTP response - """ + @blueprint.route( + '/collections//items/', + methods=['GET', 'PUT', 'DELETE', 'OPTIONS'], + provide_automatic_options=False + ) + def collection_item(collection_id: str, item_id: str) -> Response: + """ + OGC API collections item endpoint - return execute_from_flask(tiles_api.get_collection_tiles, request, - collection_id) + :param collection_id: collection identifier + :param item_id: item identifier + :returns: HTTP response + """ -@BLUEPRINT.route('/collections//tiles/') -@BLUEPRINT.route('/collections//tiles//metadata') # noqa -def get_collection_tiles_metadata(collection_id: str | None = None, - tileMatrixSetId: str | None = None): - """ - OGC open api collection tiles service metadata + if request.method == 'DELETE': + return execute_from_flask( + itemtypes_api.manage_collection_item, + request, 'delete', collection_id, item_id, + skip_valid_check=True + ) + elif request.method == 'PUT': + return execute_from_flask( + itemtypes_api.manage_collection_item, + request, 'update', collection_id, item_id, + skip_valid_check=True + ) + elif request.method == 'OPTIONS': + return execute_from_flask( + itemtypes_api.manage_collection_item, + request, 'options', collection_id, item_id, + skip_valid_check=True + ) + else: + return execute_from_flask( + itemtypes_api.get_collection_item, request, + collection_id, item_id + ) - :param collection_id: collection identifier - :param tileMatrixSetId: identifier of tile matrix set + @blueprint.route('/collections//coverage') + def collection_coverage(collection_id: str) -> Response: + """ + OGC API - Coverages coverage endpoint - :returns: HTTP response - """ + :param collection_id: collection identifier - return execute_from_flask(tiles_api.get_collection_tiles_metadata, - request, collection_id, tileMatrixSetId, - skip_valid_check=True) + :returns: HTTP response + """ + return execute_from_flask( + coverages_api.get_collection_coverage, request, + collection_id, skip_valid_check=True + ) -@BLUEPRINT.route('/collections//tiles/\ -///') -def get_collection_tiles_data(collection_id: str | None = None, - tileMatrixSetId: str | None = None, - tileMatrix: str | None = None, - tileRow: str | None = None, - tileCol: str | None = None): - """ - OGC open api collection tiles service data + @blueprint.route('/collections//tiles') + def get_collection_tiles(collection_id: str) -> Response: + """ + OGC open api collections tiles access point - :param collection_id: collection identifier - :param tileMatrixSetId: identifier of tile matrix set - :param tileMatrix: identifier of {z} matrix index - :param tileRow: identifier of {y} matrix index - :param tileCol: identifier of {x} matrix index + :param collection_id: collection identifier - :returns: HTTP response - """ + :returns: HTTP response + """ - return execute_from_flask( - tiles_api.get_collection_tiles_data, - request, collection_id, tileMatrixSetId, tileMatrix, tileRow, tileCol, - skip_valid_check=True, - ) + return execute_from_flask( + tiles_api.get_collection_tiles, request, collection_id + ) + @blueprint.route('/collections//tiles/') # noqa E501 + @blueprint.route('/collections//tiles//metadata') # noqa E501 + def get_collection_tiles_metadata( + collection_id: str, tileMatrixSetId: str) -> Response: + """ + OGC open api collection tiles service metadata -@BLUEPRINT.route('/collections//map') -@BLUEPRINT.route('/collections//styles//map') -def collection_map(collection_id: str, style_id: str | None = None): - """ - OGC API - Maps map render endpoint + :param collection_id: collection identifier + :param tileMatrixSetId: identifier of tile matrix set - :param collection_id: collection identifier - :param style_id: style identifier + :returns: HTTP response + """ - :returns: HTTP response - """ + return execute_from_flask( + tiles_api.get_collection_tiles_metadata, + request, collection_id, tileMatrixSetId, + skip_valid_check=True + ) - return execute_from_flask( - maps_api.get_collection_map, request, collection_id, style_id + @blueprint.route( + '/collections//tiles/' + '///' ) + def get_collection_tiles_data( + collection_id: str, + tileMatrixSetId: str, + tileMatrix: str, + tileRow: str, + tileCol: str + ) -> Response: + """ + OGC open api collection tiles service data + :param collection_id: collection identifier + :param tileMatrixSetId: identifier of tile matrix set + :param tileMatrix: identifier of {z} matrix index + :param tileRow: identifier of {y} matrix index + :param tileCol: identifier of {x} matrix index -@BLUEPRINT.route('/processes') -@BLUEPRINT.route('/processes/') -def get_processes(process_id: str | None = None): - """ - OGC API - Processes description endpoint - - :param process_id: process identifier - - :returns: HTTP response - """ - - return execute_from_flask(processes_api.describe_processes, request, - process_id) - - -@BLUEPRINT.route('/jobs') -@BLUEPRINT.route('/jobs/', - methods=['GET', 'DELETE']) -def get_jobs(job_id: str | None = None): - """ - OGC API - Processes jobs endpoint + :returns: HTTP response + """ - :param job_id: job identifier + return execute_from_flask( + tiles_api.get_collection_tiles_data, + request, collection_id, tileMatrixSetId, tileMatrix, + tileRow, tileCol, skip_valid_check=True + ) - :returns: HTTP response - """ + @blueprint.route('/collections//map') + @blueprint.route('/collections//styles//map') + def collection_map(collection_id: str, style_id: str = None) -> Response: + """ + OGC API - Maps map render endpoint - if job_id is None: - return execute_from_flask(processes_api.get_jobs, request) - else: - if request.method == 'DELETE': # dismiss job - return execute_from_flask(processes_api.delete_job, request, - job_id) - else: # Return status of a specific job - return execute_from_flask(processes_api.get_jobs, request, job_id) + :param collection_id: collection identifier + :param style_id: style identifier + :returns: HTTP response + """ -@BLUEPRINT.route('/processes//execution', methods=['POST']) -def execute_process_jobs(process_id: str): - """ - OGC API - Processes execution endpoint + return execute_from_flask( + maps_api.get_collection_map, request, collection_id, style_id + ) - :param process_id: process identifier + @blueprint.route('/processes') + @blueprint.route('/processes/') + def get_processes(process_id: str = None) -> Response: + """ + OGC API - Processes description endpoint - :returns: HTTP response - """ + :param process_id: process identifier - return execute_from_flask(processes_api.execute_process, request, - process_id) + :returns: HTTP response + """ + return execute_from_flask( + processes_api.describe_processes, request, process_id + ) -@BLUEPRINT.route('/jobs//results', - methods=['GET']) -def get_job_result(job_id: str | None = None): - """ - OGC API - Processes job result endpoint + @blueprint.route('/jobs') + @blueprint.route( + '/jobs/', methods=['GET', 'DELETE'] + ) + def get_jobs(job_id: str = None) -> Response: + """ + OGC API - Processes jobs endpoint - :param job_id: job identifier + :param job_id: job identifier - :returns: HTTP response - """ + :returns: HTTP response + """ - return execute_from_flask(processes_api.get_job_result, request, job_id) - - -@BLUEPRINT.route('/collections//position') -@BLUEPRINT.route('/collections//area') -@BLUEPRINT.route('/collections//cube') -@BLUEPRINT.route('/collections//radius') -@BLUEPRINT.route('/collections//trajectory') -@BLUEPRINT.route('/collections//corridor') -@BLUEPRINT.route('/collections//locations/') -@BLUEPRINT.route('/collections//locations') -@BLUEPRINT.route('/collections//instances//position') # noqa -@BLUEPRINT.route('/collections//instances//area') # noqa -@BLUEPRINT.route('/collections//instances//cube') # noqa -@BLUEPRINT.route('/collections//instances//radius') # noqa -@BLUEPRINT.route('/collections//instances//trajectory') # noqa -@BLUEPRINT.route('/collections//instances//corridor') # noqa -@BLUEPRINT.route('/collections//instances//locations/') # noqa -@BLUEPRINT.route('/collections//instances//locations') # noqa -@BLUEPRINT.route('/collections//instances/') -@BLUEPRINT.route('/collections//instances') -def get_collection_edr_query(collection_id: str, - instance_id: str | None = None, - location_id: str | None = None): - """ - OGC EDR API endpoints + if job_id is None: + return execute_from_flask(processes_api.get_jobs, request) + else: + if request.method == 'DELETE': # dismiss job + return execute_from_flask( + processes_api.delete_job, request, job_id + ) + else: # Return status of a specific job + return execute_from_flask( + processes_api.get_jobs, request, job_id + ) + + @blueprint.route('/processes//execution', methods=['POST']) + def execute_process_jobs(process_id: str) -> Response: + """ + OGC API - Processes execution endpoint - :param collection_id: collection identifier - :param instance_id: instance identifier - :param location_id: location id of a /locations/ query + :param process_id: process identifier - :returns: HTTP response - """ + :returns: HTTP response + """ - if (request.path.endswith('instances') or - (instance_id is not None and - request.path.endswith(instance_id))): return execute_from_flask( - edr_api.get_collection_edr_instances, request, collection_id, - instance_id + processes_api.execute_process, request, process_id ) - if location_id: - query_type = 'locations' - else: - query_type = request.path.split('/')[-1] - - return execute_from_flask( - edr_api.get_collection_edr_query, request, collection_id, instance_id, - query_type, location_id, skip_valid_check=True + @blueprint.route( + '/jobs//results', + methods=['GET'] ) + def get_job_result(job_id: str = None) -> Response: + """ + OGC API - Processes job result endpoint + :param job_id: job identifier -@BLUEPRINT.route('/stac-api') -def stac_landing_page(): - """ - STAC API landing page endpoint + :returns: HTTP response + """ - :returns: HTTP response - """ + return execute_from_flask( + processes_api.get_job_result, request, job_id + ) - return execute_from_flask(stac_api.landing_page, request) + @blueprint.route('/collections//position') + @blueprint.route('/collections//area') + @blueprint.route('/collections//cube') + @blueprint.route('/collections//radius') + @blueprint.route('/collections//trajectory') + @blueprint.route('/collections//corridor') + @blueprint.route('/collections//locations/') # noqa E501 + @blueprint.route('/collections//locations') + @blueprint.route('/collections//instances//position') # noqa E501 + @blueprint.route('/collections//instances//area') # noqa E501 + @blueprint.route('/collections//instances//cube') # noqa E501 + @blueprint.route('/collections//instances//radius') # noqa E501 + @blueprint.route('/collections//instances//trajectory') # noqa E501 + @blueprint.route('/collections//instances//corridor') # noqa E501 + @blueprint.route('/collections//instances//locations/') # noqa E501 + @blueprint.route('/collections//instances//locations') # noqa E501 + @blueprint.route('/collections//instances/') # noqa E501 + @blueprint.route('/collections//instances') + def get_collection_edr_query( + collection_id: str, instance_id: str = None, + location_id: str = None + ) -> Response: + """ + OGC EDR API endpoints + :param collection_id: collection identifier + :param instance_id: instance identifier + :param location_id: location id of a /locations/ query -@BLUEPRINT.route('/stac-api/search', methods=['GET', 'POST']) -def stac_search(): - """ - STAC API search endpoint + :returns: HTTP response + """ - :returns: HTTP response - """ + if ( + request.path.endswith('instances') or + (instance_id is not None and request.path.endswith(instance_id)) + ): + return execute_from_flask( + edr_api.get_collection_edr_instances, request, collection_id, + instance_id + ) - return execute_from_flask(stac_api.search, request) + if location_id: + query_type = 'locations' + else: + query_type = request.path.split('/')[-1] + return execute_from_flask( + edr_api.get_collection_edr_query, request, collection_id, + instance_id, query_type, location_id, skip_valid_check=True + ) -@BLUEPRINT.route('/stac') -def stac_catalog_root(): - """ - STAC root endpoint + @blueprint.route('/stac-api') + def stac_landing_page() -> Response: + """ + STAC API landing page endpoint - :returns: HTTP response - """ + :returns: HTTP response + """ - return execute_from_flask(stac_api.get_stac_root, request) + return execute_from_flask(stac_api.landing_page, request) + @blueprint.route('/stac-api/search', methods=['GET', 'POST']) + def stac_search() -> Response: + """ + STAC API search endpoint -@BLUEPRINT.route('/stac/') -def stac_catalog_path(path: str): - """ - STAC path endpoint + :returns: HTTP response + """ - :param path: path + return execute_from_flask(stac_api.search, request) - :returns: HTTP response - """ + @blueprint.route('/stac') + def stac_catalog_root() -> Response: + """ + STAC root endpoint - return execute_from_flask(stac_api.get_stac_path, request, path) + :returns: HTTP response + """ + return execute_from_flask(stac_api.get_stac_root, request) -@ADMIN_BLUEPRINT.route('/admin/config', methods=['GET', 'PUT', 'PATCH']) -def admin_config(): - """ - Admin endpoint + @blueprint.route('/stac/') + def stac_catalog_path(path: str) -> Response: + """ + STAC path endpoint - :returns: HTTP response - """ + :param path: path - if request.method == 'GET': - return execute_from_flask(admin_api.get_config_, request, - alternative_api=admin_) + :returns: HTTP response + """ - elif request.method == 'PUT': - return execute_from_flask(admin_api.put_config, request, - alternative_api=admin_) + return execute_from_flask(stac_api.get_stac_path, request, path) - elif request.method == 'PATCH': - return execute_from_flask(admin_api.patch_config, request, - alternative_api=admin_) + @admin_blueprint.route('/admin/config', methods=['GET', 'PUT', 'PATCH']) + def admin_config() -> Response: + """ + Admin endpoint + :returns: HTTP response + """ -@ADMIN_BLUEPRINT.route('/admin/config/resources', methods=['GET', 'POST']) -def admin_config_resources(): - """ - Resources endpoint + if request.method == 'GET': + return execute_from_flask( + admin_api.get_config_, request, alternative_api=admin_ + ) - :returns: HTTP response - """ + elif request.method == 'PUT': + return execute_from_flask( + admin_api.put_config, request, alternative_api=admin_ + ) - if request.method == 'GET': - return execute_from_flask(admin_api.get_resources, request, - alternative_api=admin_) + elif request.method == 'PATCH': + return execute_from_flask( + admin_api.patch_config, request, alternative_api=admin_ + ) - elif request.method == 'POST': - return execute_from_flask(admin_api.post_resource, request, - alternative_api=admin_) + @admin_blueprint.route('/admin/config/resources', methods=['GET', 'POST']) + def admin_config_resources() -> Response: + """ + Resources endpoint + :returns: HTTP response + """ -@ADMIN_BLUEPRINT.route( - '/admin/config/resources/', - methods=['GET', 'PUT', 'PATCH', 'DELETE']) -def admin_config_resource(resource_id: str): - """ - Resource endpoint + if request.method == 'POST': + return execute_from_flask( + admin_api.post_resource, request, alternative_api=admin_ + ) - :returns: HTTP response - """ + return execute_from_flask( + admin_api.get_resources, request, alternative_api=admin_ + ) - if request.method == 'GET': - return execute_from_flask(admin_api.get_resource, request, - resource_id, - alternative_api=admin_) + @admin_blueprint.route( + '/admin/config/resources/', + methods=['GET', 'PUT', 'PATCH', 'DELETE']) + def admin_config_resource(resource_id: str) -> Response: + """ + Resource endpoint - elif request.method == 'DELETE': - return execute_from_flask(admin_api.delete_resource, request, - resource_id, - alternative_api=admin_) + :returns: HTTP response + """ - elif request.method == 'PUT': - return execute_from_flask(admin_api.put_resource, request, - resource_id, - alternative_api=admin_) + if request.method == 'DELETE': + return execute_from_flask( + admin_api.delete_resource, request, + resource_id, alternative_api=admin_ + ) - elif request.method == 'PATCH': - return execute_from_flask(admin_api.patch_resource, request, - resource_id, - alternative_api=admin_) + elif request.method == 'PUT': + return execute_from_flask( + admin_api.put_resource, request, + resource_id, alternative_api=admin_ + ) + elif request.method == 'PATCH': + return execute_from_flask( + admin_api.patch_resource, request, + resource_id, alternative_api=admin_ + ) -APP.register_blueprint(BLUEPRINT) + # GET + return execute_from_flask( + admin_api.get_resource, request, + resource_id, alternative_api=admin_ + ) -if CONFIG['server'].get('admin'): - admin_ = Admin(CONFIG, OPENAPI) - APP.register_blueprint(ADMIN_BLUEPRINT) + app.register_blueprint(blueprint) + if config['server'].get('admin'): + admin_ = Admin(config, openapi) + app.register_blueprint(admin_blueprint) -@click.command() -@click.pass_context -@click.option('--debug', '-d', default=False, is_flag=True, help='debug') -def serve(ctx, server: str | None = None, debug: bool = False): - """ - Serve pygeoapi via Flask. Runs pygeoapi - as a flask server. Not recommend for production. + return app - :param server: `string` of server type - :param debug: `bool` of whether to run in debug mode - :returns: void - """ +if os.environ.get('PYGEOAPI_DISABLE_ENV_CONFIGS', 'false') == 'false': + APP = make_wsgi_app( + config_location=None, + openapi_location=None + ) + config = get_config() + + @click.command() + @click.pass_context + @click.option('--debug', '-d', default=False, is_flag=True, help='debug') + def serve( + ctx: click.Context, server: str = None, debug: bool = False + ) -> None: + """ + Serve pygeoapi via Flask. Runs pygeoapi + as a flask server. Not recommend for production. + + :param server: `string` of server type + :param debug: `bool` of whether to run in debug mode - # setup_logger(CONFIG['logging']) - APP.run(debug=True, host=api_.config['server']['bind']['host'], - port=api_.config['server']['bind']['port']) + :returns: void + """ + # setup_logger(CONFIG['logging']) + APP.run(debug=debug, host=config['server']['bind']['host'], + port=config['server']['bind']['port']) if __name__ == '__main__': # run locally, for testing serve() diff --git a/pygeoapi/openapi.py b/pygeoapi/openapi.py index c3016c828..5ce99f420 100644 --- a/pygeoapi/openapi.py +++ b/pygeoapi/openapi.py @@ -998,19 +998,21 @@ def generate_openapi_document(cfg_file: Union[Path, io.TextIOWrapper], return content -def load_openapi_document() -> dict: +def load_openapi_document(pygeoapi_openapi: str = None) -> dict: """ Open OpenAPI document from `PYGEOAPI_OPENAPI` environment variable + :param pygeoapi_openapi: `str` of OpenAPI document filepath, if `None` will + attempt to read from `PYGEOAPI_OPENAPI` environment + :returns: `dict` of OpenAPI document """ - - pygeoapi_openapi = os.environ.get('PYGEOAPI_OPENAPI') - if pygeoapi_openapi is None: - msg = 'PYGEOAPI_OPENAPI environment not set' - LOGGER.error(msg) - raise RuntimeError(msg) + pygeoapi_openapi = os.getenv('PYGEOAPI_OPENAPI') + if pygeoapi_openapi is None: + msg = 'PYGEOAPI_OPENAPI environment not set' + LOGGER.error(msg) + raise RuntimeError(msg) if not os.path.exists(pygeoapi_openapi): msg = (f'OpenAPI document {pygeoapi_openapi} does not exist. ' diff --git a/tests/data/mysql_data.sql b/tests/data/mysql_data.sql index f2174ae55..9bb7c34f8 100644 --- a/tests/data/mysql_data.sql +++ b/tests/data/mysql_data.sql @@ -1,4 +1,4 @@ --- A test database for the mysql provider; a simple geospatial app +-- A test database for the mysql provider; a simple geospatial app -- Create the database DROP DATABASE IF EXISTS test_geo_app; diff --git a/tests/util.py b/tests/util.py index 4cdd4373e..4431a00f1 100644 --- a/tests/util.py +++ b/tests/util.py @@ -126,7 +126,7 @@ def mock_flask(config_file: str = 'pygeoapi-test-config.yml', reload(flask_app) # Set server root path - url_parts = urlsplit(flask_app.CONFIG['server']['url']) + url_parts = urlsplit(flask_app.config['server']['url']) app_root = url_parts.path.rstrip('/') or '/' flask_app.APP.config['SERVER_NAME'] = url_parts.netloc flask_app.APP.config['APPLICATION_ROOT'] = app_root