Fix-20713: Add support for metadata ingestion using local file in REST connector (#26036)

Author: harshsoni2024

bootstrap/sql/migrations/native/1.11.11/mysql/postDataMigrationSQLScript.sql

@@ -0,0 +1,13 @@
+-- Migrate REST API service connections: move flat openAPISchemaURL into nested openAPISchemaConnection object
+UPDATE api_service_entity
+SET json = JSON_SET(
+    JSON_REMOVE(json, '$.connection.config.openAPISchemaURL'),
+    '$.connection.config.openAPISchemaConnection',
+    JSON_OBJECT(
+        'openAPISchemaURL',
+        JSON_UNQUOTE(JSON_EXTRACT(json, '$.connection.config.openAPISchemaURL'))
+    )
+)
+WHERE serviceType = 'Rest'
+  AND JSON_CONTAINS_PATH(json, 'one', '$.connection.config.openAPISchemaURL')
+  AND NOT JSON_CONTAINS_PATH(json, 'one', '$.connection.config.openAPISchemaConnection');

bootstrap/sql/migrations/native/1.11.11/mysql/schemaChanges.sql

@@ -0,0 +1,11 @@
+-- Migrate REST API service connections: move flat openAPISchemaURL into nested openAPISchemaConnection object
+UPDATE api_service_entity
+SET json = jsonb_set(
+    json #- '{connection,config,openAPISchemaURL}',
+    '{connection,config,openAPISchemaConnection}',
+    jsonb_build_object('openAPISchemaURL', json #> '{connection,config,openAPISchemaURL}'),
+    true
+)
+WHERE serviceType = 'Rest'
+  AND jsonb_exists(json -> 'connection' -> 'config', 'openAPISchemaURL')
+  AND NOT jsonb_exists(json -> 'connection' -> 'config', 'openAPISchemaConnection');

bootstrap/sql/migrations/native/1.11.11/postgres/postDataMigrationSQLScript.sql

@@ -4,7 +4,9 @@
 	"serviceConnection": {
 		"config": {
 			"type": "Rest",
-			"openAPISchemaURL": "https://petstore3.swagger.io/",
+			"openAPISchemaConnection": {
+				"openAPISchemaURL": "https://petstore3.swagger.io/"
+			},
     		"token":"mock_token"
 		}
 	},

bootstrap/sql/migrations/native/1.11.11/postgres/schemaChanges.sql

@@ -4,7 +4,9 @@
 	"serviceConnection": {
 		"config": {
 			"type": "Rest",
-			"openAPISchemaURL": "https://docs.open-metadata.org/swagger.html",
+			"openAPISchemaConnection": {
+				"openAPISchemaURL": "https://docs.open-metadata.org/swagger.html"
+			},
     		"token":"token"
 		}
 	},

ingestion/examples/sample_data/api_service/service.json

@@ -4,7 +4,10 @@ source:
   serviceConnection:
     config:
       type: Rest
-      openAPISchemaURL: https://petstore3.swagger.io/api/v3/openapi.json
+      openAPISchemaConnection:
+        openAPISchemaURL: https://petstore3.swagger.io/api/v3/openapi.json
+        # Or use a local file path instead:
+        # openAPISchemaFilePath: /path/to/openapi.json
       docURL: https://petstore3.swagger.io/
       # token: <jwt_token>
   sourceConfig:

ingestion/examples/sample_data/ometa_api_service/service.json

@@ -12,14 +12,20 @@
 """
 Source connection handler
 """
-from typing import Optional
+from typing import Dict, Optional, Union
 
 import requests
 from requests.models import Response
 
 from metadata.generated.schema.entity.automations.workflow import (
     Workflow as AutomationWorkflow,
 )
+from metadata.generated.schema.entity.services.connections.api.openAPISchemaFilePath import (
+    OpenAPISchemaFilePath,
+)
+from metadata.generated.schema.entity.services.connections.api.openAPISchemaURL import (
+    OpenAPISchemaURL,
+)
 from metadata.generated.schema.entity.services.connections.api.restConnection import (
     RestConnection,
 )
@@ -31,6 +37,7 @@
 from metadata.ingestion.source.api.rest.parser import (
     OpenAPIParseError,
     parse_openapi_schema,
+    parse_openapi_schema_from_file,
     validate_openapi_schema,
 )
 from metadata.utils.constants import THREE_MIN
@@ -48,19 +55,28 @@ class InvalidOpenAPISchemaError(Exception):
     """
 
 
-def get_connection(connection: RestConnection) -> Response:
+def get_connection(connection: RestConnection) -> Union[Response, Dict]:
     """
-    Create connection
+    Create connection.
+    If openAPISchemaURL is provided, fetches the schema via HTTP.
+    Otherwise, reads from the local openAPISchemaFilePath.
     """
-    if connection.token:
-        headers = {"Authorization": f"Bearer {connection.token.get_secret_value()}"}
-        return requests.get(connection.openAPISchemaURL, headers=headers)
-    return requests.get(connection.openAPISchemaURL)
+    schema_conn = connection.openAPISchemaConnection
+    if isinstance(schema_conn, OpenAPISchemaURL):
+        if connection.token:
+            headers = {"Authorization": f"Bearer {connection.token.get_secret_value()}"}
+            return requests.get(schema_conn.openAPISchemaURL, headers=headers)
+        return requests.get(schema_conn.openAPISchemaURL)
+
+    if isinstance(schema_conn, OpenAPISchemaFilePath):
+        return parse_openapi_schema_from_file(schema_conn.openAPISchemaFilePath)
+
+    raise ValueError(f"Unsupported openAPISchemaConnection type: {type(schema_conn)}")
 
 
 def test_connection(
     metadata: OpenMetadata,
-    client: Response,
+    client: Union[Response, Dict],
     service_connection: RestConnection,
     automation_workflow: Optional[AutomationWorkflow] = None,
     timeout_seconds: Optional[int] = THREE_MIN,
@@ -69,18 +85,22 @@ def test_connection(
     Test connection. This can be executed either as part
     of a metadata workflow or during an Automation Workflow
     """
+    is_local_file = isinstance(client, dict)
 
     def custom_url_exec():
+        if is_local_file:
+            return []
         if client.status_code == 200:
             return []
         raise SchemaURLError(
-            "Failed to connect to the JSON schema url. Please check the url and credentials. Status Code was: "
-            + str(client.status_code)
+            "Failed to connect to the JSON schema url. "
+            "Please check the url and credentials. "
+            f"Status Code was: {client.status_code}"
         )
 
     def custom_schema_exec():
         try:
-            schema = parse_openapi_schema(client)
+            schema = client if is_local_file else parse_openapi_schema(client)
             if validate_openapi_schema(schema):
                 return []
 
@@ -89,6 +109,8 @@ def custom_schema_exec():
             )
         except OpenAPIParseError as e:
             raise InvalidOpenAPISchemaError(f"Failed to parse OpenAPI schema: {e}")
+        except InvalidOpenAPISchemaError:
+            raise
         except Exception as e:
             raise InvalidOpenAPISchemaError(f"Error validating OpenAPI schema: {e}")
 

ingestion/src/metadata/examples/workflows/rest.yaml

@@ -23,6 +23,9 @@
 )
 from metadata.generated.schema.entity.data.apiCollection import APICollection
 from metadata.generated.schema.entity.data.apiEndpoint import ApiRequestMethod
+from metadata.generated.schema.entity.services.connections.api.openAPISchemaURL import (
+    OpenAPISchemaURL,
+)
 from metadata.generated.schema.entity.services.connections.api.restConnection import (
     RestConnection,
 )
@@ -80,7 +83,10 @@ def get_api_collections(self, *args, **kwargs) -> Iterable[RESTCollection]:
         Here is where filtering happens
         """
         try:
-            self.json_response = parse_openapi_schema(self.connection)
+            if isinstance(self.connection, dict):
+                self.json_response = self.connection
+            else:
+                self.json_response = parse_openapi_schema(self.connection)
             collections_list = []
             tags_collection_set = set()
             if self.json_response.get("tags", []):
@@ -222,6 +228,13 @@ def _prepare_endpoint_data(
             logger.warning(f"Error while parsing endpoint data: {err}")
         return None
 
+    def _get_fallback_url(self) -> Optional[AnyUrl]:
+        """Return openAPISchemaURL if available, otherwise None."""
+        schema_conn = self.config.serviceConnection.root.config.openAPISchemaConnection
+        if isinstance(schema_conn, OpenAPISchemaURL):
+            return schema_conn.openAPISchemaURL
+        return None
+
     def _generate_collection_url(self, collection_name: str) -> Optional[AnyUrl]:
         """generate collection url"""
         try:
@@ -231,7 +244,7 @@ def _generate_collection_url(self, collection_name: str) -> Optional[AnyUrl]:
                     f"Could not generate collection url for {collection_name}"
                     " because docURL is not present"
                 )
-                return self.config.serviceConnection.root.config.openAPISchemaURL
+                return self._get_fallback_url()
             base_url = str(base_url)
             if base_url.endswith("#/") or base_url.endswith("#"):
                 base_url = base_url.split("#")[0]
@@ -240,11 +253,11 @@ def _generate_collection_url(self, collection_name: str) -> Optional[AnyUrl]:
             logger.warning(
                 f"Error while generating collection url for {collection_name}: {err}"
             )
-        return self.config.serviceConnection.root.config.openAPISchemaURL
+        return self._get_fallback_url()
 
     def _generate_endpoint_url(
         self, collection: RESTCollection, endpoint: RESTEndpoint
-    ) -> AnyUrl:
+    ) -> Optional[AnyUrl]:
         """generate endpoint url"""
         try:
             if not collection.url or not endpoint.operationId:
@@ -253,11 +266,11 @@ def _generate_endpoint_url(
                     f" collection url: {str(collection.url)},"
                     f" endpoint operation id: {str(endpoint.operationId)}"
                 )
-                return self.config.serviceConnection.root.config.openAPISchemaURL
+                return self._get_fallback_url()
             return AnyUrl(f"{str(collection.url)}/{endpoint.operationId}")
         except Exception as err:
             logger.warning(f"Error while generating collection url: {err}")
-        return self.config.serviceConnection.root.config.openAPISchemaURL
+        return self._get_fallback_url()
 
     def _get_api_request_method(self, method_type: str) -> Optional[str]:
         """fetch endpoint request method"""

ingestion/src/metadata/ingestion/source/api/rest/connection.py

@@ -13,7 +13,8 @@
 OpenAPI schema parser for both JSON and YAML formats
 """
 import json
-from typing import Any, Dict
+from pathlib import Path
+from typing import Any, Dict, Union
 
 import yaml
 from requests.models import Response
@@ -101,3 +102,49 @@ def validate_openapi_schema(schema: Dict[str, Any]) -> bool:
 
     # OpenAPI 3.x uses "openapi" field, OpenAPI 2.x uses "swagger" field
     return schema.get("openapi") is not None or schema.get("swagger") is not None
+
+
+def parse_openapi_schema_from_file(file_path: Union[str, Path]) -> Dict[str, Any]:
+    """
+    Parse OpenAPI schema from a local file.
+    Supports both JSON and YAML formats.
+    """
+    path = Path(file_path)
+    if not path.exists():
+        raise OpenAPIParseError(f"File not found: {file_path}")
+    if not path.is_file():
+        raise OpenAPIParseError(f"Path is not a file: {file_path}")
+
+    content = path.read_text(encoding="utf-8")
+    suffix = path.suffix.lower()
+
+    if suffix in (".json",):
+        try:
+            return json.loads(content)
+        except json.JSONDecodeError as e:
+            raise OpenAPIParseError(f"Failed to parse JSON file: {e}") from e
+
+    if suffix in (".yaml", ".yml"):
+        try:
+            parsed = yaml.safe_load(content)
+            if parsed is None:
+                raise OpenAPIParseError("YAML parsing returned None")
+            return parsed
+        except yaml.YAMLError as e:
+            raise OpenAPIParseError(f"Failed to parse YAML file: {e}") from e
+
+    # Unknown extension — try JSON first, then YAML
+    try:
+        return json.loads(content)
+    except json.JSONDecodeError:
+        pass
+
+    try:
+        parsed = yaml.safe_load(content)
+        if parsed is None:
+            raise OpenAPIParseError("YAML parsing returned None")
+        return parsed
+    except yaml.YAMLError:
+        pass
+
+    raise OpenAPIParseError(f"Failed to parse '{file_path}' as either JSON or YAML.")