feat: separate a function for adding external Swagger to an openapi.py

Signed-off-by: ImMin5 <mino@megazone.com>

Author: ImMin5

src/spaceone/core/fastapi/openapi.py

@@ -0,0 +1,122 @@
+import logging
+import json
+import os
+
+from fastapi import FastAPI
+from fastapi.openapi.docs import get_swagger_ui_html
+from fastapi.responses import HTMLResponse
+
+from spaceone.core import config
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def _add_external_api_route(
+        app: FastAPI,
+        prefix: str,
+        service_name: str,
+        external_swagger_path: str
+) -> None:
+    async def get_openapi():
+        openapi_json_file = os.path.join(external_swagger_path, f"{service_name}_openapi.json")
+        with open(openapi_json_file, 'r') as f:
+            custom_openapi_schema = json.loads(f.read())
+        return custom_openapi_schema
+
+    get_openapi.__name__ = get_openapi.__name__ + prefix
+    app.add_api_route(prefix + "/openapi.json", get_openapi, include_in_schema=False)
+
+    async def swagger_ui_html() -> HTMLResponse:
+        return get_swagger_ui_html(
+            openapi_url=prefix + "/openapi.json",
+            title=f'{service_name.title().replace("-", " ")} API' + ' - Swagger UI',
+            oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url,
+            init_oauth=app.swagger_ui_init_oauth,
+        )
+
+    swagger_ui_html.__name__ = swagger_ui_html.__name__ + prefix
+    app.add_api_route(prefix + "/docs", swagger_ui_html, include_in_schema=False)
+
+
+def _services_from_openapi_json_files(openapi_json_path):
+    """
+    openapi json file must follow {service}_openapi.json file name format.
+    if '-' exist in {service} converted to '_'.
+    """
+    services = []
+    openapi_json_files = os.listdir(openapi_json_path)
+    for openapi_json_file in openapi_json_files:
+        services.append('_'.join(openapi_json_file.split('_')[:-1]).lower())
+    return services
+
+
+def _sort_services(services):
+    external_swagger_priority_services = config.get_global('EXTERNAL_SWAGGER_PRIORITY_SERVICES', [])
+    external_swagger_priority_services = [external_service.replace('-', '_').lower() for external_service in
+                                          external_swagger_priority_services]
+
+    sorted_services = list(dict.fromkeys(external_swagger_priority_services + sorted(services)))
+    return sorted_services
+
+
+def _create_openapi_json(app, service_name, external_swagger_path, service_description):
+    openapi_json_file = os.path.join(external_swagger_path, f"{service_name}_openapi.json")
+    service_name_with_dash = service_name.replace('_', '-')
+    try:
+        with open(openapi_json_file, 'r') as f:
+            custom_openapi_schema = json.loads(f.read())
+            custom_openapi_schema['openapi'] = app.openapi().get('openapi')
+            description = custom_openapi_schema['info']['summary']
+            service_description[service_name] = f"| **{service_name.replace('_', ' ').title()}** | {description} | [/{service_name_with_dash}/docs](/{service_name_with_dash}/docs) |\n"
+
+        with open(openapi_json_file, 'w') as f:
+            json.dump(custom_openapi_schema, f, indent=2)
+    except Exception as e:
+        _LOGGER.error(f'[_create_openapi_json] {openapi_json_file} : {e}', exc_info=True)
+
+
+def _check_external_swagger_path(external_swagger_path):
+    if not external_swagger_path:
+        _LOGGER.info('[_check_external_swagger_path] EXTERNAL_SWAGGER_PATH is not set')
+        return False
+
+    if not os.path.exists(external_swagger_path):
+        _LOGGER.error(f'[_check_external_swagger_path] "{external_swagger_path}" : Not Found')
+        return False
+
+    return True
+
+
+def _create_external_apis_description(app, service_description):
+    app.openapi()['info']['description'] += "\n<br><br>\n"
+    app.openapi()['info']['description'] += "\n## List of External APIs\n"
+    app.openapi()['info']['description'] += "\n[Home](/docs)\n"
+    app.openapi()['info']['description'] += "| **Name** | **Description** | **URL** |\n"
+    app.openapi()['info']['description'] += "|:---|:--- |:---|\n"
+    app.openapi()['info']['description'] += ''.join(service_description.values())
+
+
+def add_external_swagger(app):
+    try:
+        external_swagger_path = config.get_global('EXTERNAL_SWAGGER_PATH')
+
+        if not _check_external_swagger_path(external_swagger_path):
+            return app
+
+        services = _services_from_openapi_json_files(external_swagger_path)
+        sorted_services = _sort_services(services)
+
+        service_description = {}
+        for service in sorted_services:
+            _create_openapi_json(app=app, service_name=service, external_swagger_path=external_swagger_path,
+                                 service_description=service_description)
+            _add_external_api_route(app=app, prefix=f"/{service.replace('_', '-')}", service_name=service,
+                                    external_swagger_path=external_swagger_path)
+
+        if len(services) > 0:
+            _create_external_apis_description(app, service_description)
+
+    except Exception as e:
+        _LOGGER.error(f'[add_external_swagger] {e}', exc_info=True)
+    finally:
+        return app

src/spaceone/core/fastapi/server.py

@@ -1,17 +1,14 @@
 import logging
 import uvicorn
-import json
-import os
 
-from fastapi import FastAPI, Request
+from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.openapi.docs import get_swagger_ui_html
-from fastapi.responses import HTMLResponse
 
 from spaceone.core import config
 from spaceone.core.logger import set_logger
 from spaceone.core.opentelemetry import set_tracer, set_metric
 from spaceone.core.extension.server_info import ServerInfoManager
+from spaceone.core.fastapi.openapi import add_external_swagger
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -134,107 +131,11 @@ def _init_fast_api():
     )
 
 
-def _get_all_services_from_openapi_json_files(openapi_json_path):
-    services = []
-    openapi_json_files = os.listdir(openapi_json_path)
-    for openapi_json_file in openapi_json_files:
-        services.append('_'.join(openapi_json_file.split('_')[:-1]).lower())
-    return services
-
-
-def _sort_services(services):
-    return sorted(services, key=lambda x: ('identity' not in x, 'inventory' not in x, 'cost-analysis' not in x,
-                                           'monitoring' not in x, 'notification' not in x, 'repository' not in x, x))
-
-
-def _create_openapi_json(app, service_name):
-    swagger_path = config.get_global('EXTENSION_SWAGGER_PATH')
-    swagger_path = os.path.join(swagger_path, f"{service_name.replace('-', '_')}_openapi.json")
-    try:
-        with open(swagger_path, 'r') as f:
-            custom_openapi_schema = json.loads(f.read())
-            custom_openapi_schema['openapi'] = app.openapi().get('openapi')
-            description = custom_openapi_schema['info']['summary']
-            app.openapi()['info']['description'] += f"| **{service_name.replace('-', ' ').title()}** | {description} | [/{service_name}/docs](/{service_name}/docs) |\n"
-
-        with open(swagger_path, 'w') as f:
-            json.dump(custom_openapi_schema, f, indent=2)
-    except Exception as e:
-        _LOGGER.error(f'[_create_openapi_json] {swagger_path} : {e}', exc_info=True)
-
-
-def _override_openapi(app):
-    extension_swagger_path = config.get_global('EXTENSION_SWAGGER_PATH')
-    if not os.path.exists(extension_swagger_path):
-        _LOGGER.info(f'[_override_openapi] Extension Swagger Path is not exists. (path = {extension_swagger_path})')
-        return app
-
-    services = _get_all_services_from_openapi_json_files(extension_swagger_path)
-    services = _sort_services(services)
-    _openapi_info = app.openapi().get('info')
-    _openapi_version = app.openapi().get('openapi')
-
-    app.openapi()['info']['description'] += "\n<br><br>\n"
-    app.openapi()['info']['description'] += "\n## List of Services\n"
-    app.openapi()['info']['description'] += "\n[Home](/docs)\n"
-    app.openapi()['info']['description'] += "| **Service** | **Description** | **URL** |\n"
-    app.openapi()['info']['description'] += "|:---|:--- |:---|\n"
-
-    for service in services:
-        service = service.replace('_', '-')
-        _create_openapi_json(app, service_name=service)
-        build_docs(app, prefix=f"/{service}", service_name=service)
-
-    app.openapi()['info'][
-        'description'] += "| **Console API** | Service that offers features exclusive to the Console API. | [/docs](/docs#console-api%20%3E%20api) |\n"
-
-    return app
-
-
-def build_docs(
-    app: FastAPI,
-    prefix: str,
-    service_name: str
-) -> None:
-    async def get_openapi():
-        swagger_path = config.get_global('EXTENSION_SWAGGER_PATH')
-        swagger_path = os.path.join(swagger_path, f"{service_name.replace('-','_')}_openapi.json")
-        with open(swagger_path, 'r') as f:
-            custom_openapi_schema = json.loads(f.read())
-        return custom_openapi_schema
-
-    get_openapi.__name__ = get_openapi.__name__ + prefix
-    app.add_api_route(prefix + "/openapi.json", get_openapi, include_in_schema=False)
-
-    async def swagger_ui_html() -> HTMLResponse:
-        return get_swagger_ui_html(
-            openapi_url=prefix + "/openapi.json",
-            title=f'{service_name.title().replace("-"," ")} API' + ' - Swagger UI',
-            oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url,
-            init_oauth=app.swagger_ui_init_oauth,
-        )
-
-    swagger_ui_html.__name__ = swagger_ui_html.__name__ + prefix
-    app.add_api_route(prefix + "/docs", swagger_ui_html, include_in_schema=False)
-
-
-def _get_all_services_list(app):
-    services = []
-    for route in app.routes:
-        path = route.path.split('/')
-        if len(path) == 4:
-            services.append(path[1].replace('-', '_'))
-
-    services = list(set(services))
-    sorted_services = sorted(services, key=lambda x: ('identity' not in x, 'inventory' not in x, 'cost_analysis' not in x, x))
-    return sorted_services
-
-
 def fast_api_app():
     app = _init_fast_api()
     app = _add_middlewares(app)
     app = _include_routers(app)
-    app = _override_openapi(app)
+    app = add_external_swagger(app)
     return app