add OpenAPI dict to pygeoapi.api.API init (#1398)

pygeoapi/api.py

@@ -641,16 +641,18 @@ class APIRequest:
 class API:
     """API object"""
 
-    def __init__(self, config):
+    def __init__(self, config, openapi):
         """
         constructor
 
         :param config: configuration dict
+        :param openapi: openapi dict
 
         :returns: `pygeoapi.API` instance
         """
 
         self.config = config
+        self.openapi = openapi
         self.api_headers = get_api_rules(self.config).response_headers
         self.base_url = get_base_url(self.config)
         self.prefetcher = UrlPrefetcher()
@@ -790,8 +792,8 @@ class API:
 
     @gzip
     @pre_process
-    def openapi(self, request: Union[APIRequest, Any],
+    def openapi_(self, request: Union[APIRequest, Any]) -> Tuple[
-                openapi) -> Tuple[dict, int, str]:
+                 dict, int, str]:
         """
         Provide OpenAPI document
 
@@ -821,10 +823,11 @@ class API:
 
         headers['Content-Type'] = 'application/vnd.oai.openapi+json;version=3.0'  # noqa
 
-        if isinstance(openapi, dict):
+        if isinstance(self.openapi, dict):
-            return headers, HTTPStatus.OK, to_json(openapi, self.pretty_print)
+            return headers, HTTPStatus.OK, to_json(self.openapi,
+                                                   self.pretty_print)
         else:
-            return headers, HTTPStatus.OK, openapi
+            return headers, HTTPStatus.OK, self.openapi
 
     @gzip
     @pre_process

pygeoapi/django_/views.py

@@ -38,7 +38,6 @@ from typing import Tuple, Dict, Mapping, Optional
 from django.conf import settings
 from django.http import HttpRequest, HttpResponse
 from pygeoapi.api import API
-from pygeoapi.openapi import get_oas
 
 
 def landing_page(request: HttpRequest) -> HttpResponse:
@@ -65,8 +64,7 @@ def openapi(request: HttpRequest) -> HttpResponse:
     :returns: Django HTTP Response
     """
 
-    openapi_config = get_oas(settings.PYGEOAPI_CONFIG)
+    response_ = _feed_response(request, 'openapi_')
-    response_ = _feed_response(request, 'openapi', openapi_config)
     response = _to_django_response(*response_)
 
     return response

pygeoapi/flask_app.py

@@ -37,6 +37,7 @@ import click
 from flask import Flask, Blueprint, make_response, request, send_from_directory
 
 from pygeoapi.api import API
+from pygeoapi.openapi import load_openapi_document
 from pygeoapi.util import get_mimetype, yaml_load, get_api_rules
 
 
@@ -46,6 +47,11 @@ if 'PYGEOAPI_CONFIG' not in os.environ:
 with open(os.environ.get('PYGEOAPI_CONFIG'), encoding='utf8') as fh:
     CONFIG = yaml_load(fh)
 
+if 'PYGEOAPI_OPENAPI' not in os.environ:
+    raise RuntimeError('PYGEOAPI_OPENAPI environment variable not set')
+
+OPENAPI = load_openapi_document()
+
 API_RULES = get_api_rules(CONFIG)
 
 STATIC_FOLDER = 'static'
@@ -73,7 +79,7 @@ if CONFIG['server'].get('cors', False):
 APP.config['JSONIFY_PRETTYPRINT_REGULAR'] = CONFIG['server'].get(
     'pretty_print', True)
 
-api_ = API(CONFIG)
+api_ = API(CONFIG, OPENAPI)
 
 OGC_SCHEMAS_LOCATION = CONFIG['server'].get('ogc_schemas_location')
 
@@ -139,13 +145,7 @@ def openapi():
 
     :returns: HTTP response
     """
-    with open(os.environ.get('PYGEOAPI_OPENAPI'), encoding='utf8') as ff:
+    return get_response(api_.openapi_(request))
-        if os.environ.get('PYGEOAPI_OPENAPI').endswith(('.yaml', '.yml')):
-            openapi_ = yaml_load(ff)
-        else:  # JSON string, do not transform
-            openapi_ = ff.read()
-
-    return get_response(api_.openapi(request, openapi_))
 
 
 @BLUEPRINT.route('/conformance')

pygeoapi/openapi.py

@@ -4,7 +4,7 @@
 # Authors: Francesco Bartoli <xbartolone@gmail.com>
 # Authors: Ricardo Garcia Silva <ricardo.garcia.silva@geobeyond.it>
 #
-# Copyright (c) 2022 Tom Kralidis
+# Copyright (c) 2023 Tom Kralidis
 # Copyright (c) 2022 Francesco Bartoli
 # Copyright (c) 2023 Ricardo Garcia Silva
 #
@@ -1364,6 +1364,24 @@ def generate_openapi_document(cfg_file: Union[Path, io.TextIOWrapper],
     return content
 
 
+def load_openapi_document() -> dict:
+    """
+    Open OpenAPI document from `PYGEOAPI_OPENAPI` environment variable
+
+    :returns: `dict` of OpenAPI document
+    """
+
+    pygeoapi_openapi = os.environ.get('PYGEOAPI_OPENAPI')
+
+    with open(pygeoapi_openapi, encoding='utf8') as ff:
+        if pygeoapi_openapi.endswith(('.yaml', '.yml')):
+            openapi_ = yaml_load(ff)
+        else:  # JSON string, do not transform
+            openapi_ = ff.read()
+
+    return openapi_
+
+
 @click.group()
 def openapi():
     """OpenAPI management"""

pygeoapi/starlette_app.py

@@ -50,6 +50,7 @@ from starlette.responses import (
 import uvicorn
 
 from pygeoapi.api import API
+from pygeoapi.openapi import load_openapi_document
 from pygeoapi.util import yaml_load, get_api_rules
 
 if 'PYGEOAPI_CONFIG' not in os.environ:
@@ -58,6 +59,11 @@ if 'PYGEOAPI_CONFIG' not in os.environ:
 with open(os.environ.get('PYGEOAPI_CONFIG'), encoding='utf8') as fh:
     CONFIG = yaml_load(fh)
 
+if 'PYGEOAPI_OPENAPI' not in os.environ:
+    raise RuntimeError('PYGEOAPI_OPENAPI environment variable not set')
+
+OPENAPI = load_openapi_document()
+
 p = Path(__file__)
 
 APP = Starlette(debug=True)
@@ -70,7 +76,7 @@ except KeyError:
 
 API_RULES = get_api_rules(CONFIG)
 
-api_ = API(CONFIG)
+api_ = API(CONFIG, OPENAPI)
 
 
 def get_response(result: tuple) -> Union[Response, JSONResponse, HTMLResponse]:
@@ -116,13 +122,7 @@ async def openapi(request: Request):
 
     :returns: Starlette HTTP Response
     """
-    with open(os.environ.get('PYGEOAPI_OPENAPI'), encoding='utf8') as ff:
+    return get_response(api_.openapi_(request))
-        if os.environ.get('PYGEOAPI_OPENAPI').endswith(('.yaml', '.yml')):
-            openapi_ = yaml_load(ff)
-        else:  # JSON file, do not transform
-            openapi_ = ff
-
-    return get_response(api_.openapi(request, openapi_))
 
 
 async def conformance(request: Request):

pytest.ini

@@ -1,4 +1,4 @@
 [pytest]
 env =
     PYGEOAPI_CONFIG=pygeoapi-config.yml
-    PYGEOAPI_OPENAPI=pygeoapi-openapi.yml
+    PYGEOAPI_OPENAPI=tests/pygeoapi-test-openapi.yml

tests/test_api.py

@@ -89,14 +89,14 @@ def openapi():
 
 
 @pytest.fixture()
-def api_(config):
+def api_(config, openapi):
-    return API(config)
+    return API(config, openapi)
 
 
 @pytest.fixture()
 def enclosure_api(config_enclosure):
     """ Returns an API instance with a collection with enclosure links. """
-    return API(config_enclosure)
+    return API(config_enclosure, openapi)
 
 
 @pytest.fixture()
@@ -104,12 +104,12 @@ def rules_api(config_with_rules):
     """ Returns an API instance with URL prefix and strict slashes policy.
     The API version is extracted from the current version here.
     """
-    return API(config_with_rules)
+    return API(config_with_rules, openapi)
 
 
 @pytest.fixture()
 def api_hidden_resources(config_hidden_resources):
-    return API(config_hidden_resources)
+    return API(config_hidden_resources, openapi)
 
 
 def test_apirequest(api_):
@@ -376,7 +376,7 @@ def test_api(config, api_, openapi):
     assert isinstance(api_.config, dict)
 
     req = mock_request(HTTP_ACCEPT='application/json')
-    rsp_headers, code, response = api_.openapi(req, openapi)
+    rsp_headers, code, response = api_.openapi_(req)
     assert rsp_headers['Content-Type'] == 'application/vnd.oai.openapi+json;version=3.0'  # noqa
     # No language requested: should be set to default from YAML
     assert rsp_headers['Content-Language'] == 'en-US'
@@ -385,7 +385,7 @@ def test_api(config, api_, openapi):
 
     a = 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8'
     req = mock_request(HTTP_ACCEPT=a)
-    rsp_headers, code, response = api_.openapi(req, openapi)
+    rsp_headers, code, response = api_.openapi_(req)
     assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] == \
            FORMAT_TYPES[F_HTML]
 
@@ -393,14 +393,14 @@ def test_api(config, api_, openapi):
 
     a = 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8'
     req = mock_request({'ui': 'redoc'}, HTTP_ACCEPT=a)
-    rsp_headers, code, response = api_.openapi(req, openapi)
+    rsp_headers, code, response = api_.openapi_(req)
     assert rsp_headers['Content-Type'] == FORMAT_TYPES[F_HTML] == \
            FORMAT_TYPES[F_HTML]
 
     assert 'ReDoc' in response
 
     req = mock_request({'f': 'foo'})
-    rsp_headers, code, response = api_.openapi(req, openapi)
+    rsp_headers, code, response = api_.openapi_(req)
     assert rsp_headers['Content-Language'] == 'en-US'
     assert code == HTTPStatus.BAD_REQUEST
 
@@ -445,7 +445,7 @@ def test_gzip(config, api_):
     config['server']['gzip'] = True
     enc_16 = 'utf-16'
     config['server']['encoding'] = enc_16
-    api_ = API(config)
+    api_ = API(config, openapi)
 
     # Responses from server with gzip compression
     rsp_json_headers, _, rsp_gzip_json = api_.landing_page(req_gzip_json)
@@ -529,7 +529,7 @@ def test_gzip_csv(config, api_):
 
     # Use utf-16 encoding
     config['server']['encoding'] = 'utf-16'
-    api_ = API(config)
+    api_ = API(config, openapi)
 
     req_csv = mock_request({'f': 'csv'}, HTTP_ACCEPT_ENCODING=F_GZIP)
     rsp_csv_headers, _, rsp_csv_gzip = api_.get_collection_items(req_csv, 'obs') # noqa
@@ -1183,7 +1183,7 @@ def test_manage_collection_item_editable_options_req(config):
     """Test OPTIONS request on a editable items endpoint"""
     config = copy.deepcopy(config)
     config['resources']['obs']['providers'][0]['editable'] = True
-    api_ = API(config)
+    api_ = API(config, openapi)
 
     req = mock_request()
     rsp_headers, code, _ = api_.manage_collection_item(req, 'options', 'obs')

tests/test_api_ogr_provider.py

@@ -31,10 +31,11 @@
 
 import json
 import logging
-import pytest
-from pygeoapi.api import (API)
-from pygeoapi.util import yaml_load, geojson_to_geom
 
+import pytest
+
+from pygeoapi.api import API
+from pygeoapi.util import yaml_load, geojson_to_geom
 from .util import get_test_file_path, mock_request
 
 LOGGER = logging.getLogger(__name__)
@@ -48,9 +49,15 @@ def config():
         return yaml_load(fh)
 
 
+@pytest.fixture()
+def openapi():
+    with open(get_test_file_path('pygeoapi-test-openapi.yml')) as fh:
+        return yaml_load(fh)
+
+
 @pytest.fixture()
 def api_(config):
-    return API(config)
+    return API(config, openapi)
 
 
 def test_get_collection_items_bbox_crs(config, api_):

tests/test_postgresql_provider.py

@@ -85,12 +85,18 @@ def config():
     }
 
 
+@pytest.fixture()
+def openapi():
+    with open(get_test_file_path('pygeoapi-test-openapi.yml')) as fh:
+        return yaml_load(fh)
+
+
 # API using PostgreSQL provider
 @pytest.fixture()
-def pg_api_():
+def pg_api_(openapi):
     with open(get_test_file_path('pygeoapi-test-config-postgresql.yml')) as fh:
         config = yaml_load(fh)
-        return API(config)
+        return API(config, openapi)
 
 
 def test_valid_connection_options(config):

tests/test_tinydb_manager_for_parallel_requests.py

@@ -29,21 +29,17 @@
 #
 # =================================================================
 
+import json
 from pathlib import Path
 
+from multiprocessing import Process, Manager
+import pytest
+from tinydb import TinyDB, Query
 from werkzeug.wrappers import Request
 from werkzeug.test import create_environ
-from multiprocessing import Process, Manager
-import json
 
-from tinydb import TinyDB, Query
+from pygeoapi.api import API, APIRequest
-
-import pytest
-from pygeoapi.api import (
-    API, APIRequest
-)
 from pygeoapi.util import yaml_load
-
 from .util import get_test_file_path
 
 
@@ -54,8 +50,14 @@ def config():
 
 
 @pytest.fixture()
-def api_(config):
+def openapi():
-    return API(config)
+    with open(get_test_file_path('pygeoapi-test-openapi.yml')) as fh:
+        return yaml_load(fh)
+
+
+@pytest.fixture()
+def api_(config, openapi):
+    return API(config, openapi)
 
 
 def _execute_process(api, request, process_id, index, processes_out):

tests/util.py

@@ -78,7 +78,9 @@ def mock_request(params: dict = None, data=None, **headers) -> Request:
 
 
 @contextmanager
-def mock_flask(config_file: str = 'pygeoapi-test-config.yml', **kwargs) -> FlaskClient:  # noqa
+def mock_flask(config_file: str = 'pygeoapi-test-config.yml',
+               openapi_file: str = 'pygeoapi-test-openapi.yml',
+               **kwargs) -> FlaskClient:
     """
     Mocks a Flask client so we can test the API routing with applied API rules.
     Does not follow redirects by default. Set `follow_redirects=True` option
@@ -86,12 +88,16 @@ def mock_flask(config_file: str = 'pygeoapi-test-config.yml', **kwargs) -> Flask
 
     :param config_file: Optional configuration YAML file to use.
                         If not set, the default test configuration is used.
+
+    :param openapi_file: Optional OpenAPI YAML file to use.
     """
     flask_app = None
     env_conf = os.getenv('PYGEOAPI_CONFIG')
+    env_openapi = os.getenv('PYGEOAPI_OPENAPI')
     try:
         # Temporarily override environment variable so we can import Flask app
         os.environ['PYGEOAPI_CONFIG'] = get_test_file_path(config_file)
+        os.environ['PYGEOAPI_OPENAPI'] = get_test_file_path(openapi_file)
 
         # Import current pygeoapi Flask app module
         from pygeoapi import flask_app
@@ -110,21 +116,25 @@ def mock_flask(config_file: str = 'pygeoapi-test-config.yml', **kwargs) -> Flask
         yield client
 
     finally:
-        if env_conf is None:
+        if env_conf is None and env_openapi is None:
             # Remove env variable again if it was not set initially
             del os.environ['PYGEOAPI_CONFIG']
+            del os.environ['PYGEOAPI_OPENAPI']
             # Unload Flask app module
             del sys.modules['pygeoapi.flask_app']
         else:
             # Restore env variable to its original value and reload Flask app
             os.environ['PYGEOAPI_CONFIG'] = env_conf
+            os.environ['PYGEOAPI_OPENAPI'] = env_openapi
             if flask_app:
                 reload(flask_app)
         del client
 
 
 @contextmanager
-def mock_starlette(config_file: str = 'pygeoapi-test-config.yml', **kwargs) -> StarletteClient:  # noqa
+def mock_starlette(config_file: str = 'pygeoapi-test-config.yml',
+                   openapi_file: str = 'pygeoapi-test-openapi.yml',
+                   **kwargs) -> StarletteClient:
     """
     Mocks a Starlette client so we can test the API routing with applied
     API rules.
@@ -133,12 +143,17 @@ def mock_starlette(config_file: str = 'pygeoapi-test-config.yml', **kwargs) -> S
 
     :param config_file: Optional configuration YAML file to use.
                         If not set, the default test configuration is used.
+
+    :param openapi_file: Optional OpenAPI YAML file to use.
     """
+
     starlette_app = None
     env_conf = os.getenv('PYGEOAPI_CONFIG')
+    env_openapi = os.getenv('PYGEOAPI_OPENAPI')
     try:
         # Temporarily override environment variable to import Starlette app
         os.environ['PYGEOAPI_CONFIG'] = get_test_file_path(config_file)
+        os.environ['PYGEOAPI_OPENAPI'] = get_test_file_path(openapi_file)
 
         # Import current pygeoapi Starlette app module
         from pygeoapi import starlette_app
@@ -164,14 +179,16 @@ def mock_starlette(config_file: str = 'pygeoapi-test-config.yml', **kwargs) -> S
         yield client
 
     finally:
-        if env_conf is None:
+        if env_conf is None and env_openapi is None:
             # Remove env variable again if it was not set initially
             del os.environ['PYGEOAPI_CONFIG']
+            del os.environ['PYGEOAPI_OPENAPI']
             # Unload Starlette app module
             del sys.modules['pygeoapi.starlette_app']
         else:
             # Restore env variable to original value and reload Starlette app
             os.environ['PYGEOAPI_CONFIG'] = env_conf
+            os.environ['PYGEOAPI_OPENAPI'] = env_openapi
             if starlette_app:
                 reload(starlette_app)
         del client