add better exceptions for broken spec files

Author: sh-rp

dlt_init_openapi/cli/__init__.py

@@ -9,6 +9,7 @@
 
 from dlt_init_openapi.cli.cli_endpoint_selection import questionary_endpoint_selection
 from dlt_init_openapi.config import Config
+from dlt_init_openapi.exceptions import DltOpenAPITerminalException
 from dlt_init_openapi.utils import update_rest_api
 
 app = typer.Typer(add_completion=False)
@@ -91,31 +92,38 @@ def _init_command_wrapped(
         typer.secho("Provide either --url or --path, not both", fg=typer.colors.RED)
         raise typer.Exit(code=1)
 
-    # synch rest api
-    update_rest_api.update_rest_api(force=update_rest_api_source)
-
-    config = _load_config(
-        path=config_path,
-        config={
-            "project_name": source,
-            "package_name": source,
-            "output_path": output_path,
-            "endpoint_filter": questionary_endpoint_selection if interactive else None,
-            "global_limit": global_limit,
-        },
-    )
-
-    if config.project_dir.exists():
-        if not questionary.confirm(
-            f"Directory {config.project_dir} exists, do you want to continue and update the generated files? "
-            + "This will overwrite your changes in those files."
-        ).ask():
-            logger.warning("Exiting...")
-            exit(0)
-
-    create_new_client(
-        url=url,
-        path=path,
-        config=config,
-    )
-    logger.success("Pipeline created. Learn more at https://dlthub.com/docs. See you next time :)")
+    try:
+
+        # synch rest api
+        update_rest_api.update_rest_api(force=update_rest_api_source)
+
+        config = _load_config(
+            path=config_path,
+            config={
+                "project_name": source,
+                "package_name": source,
+                "output_path": output_path,
+                "endpoint_filter": questionary_endpoint_selection if interactive else None,
+                "global_limit": global_limit,
+            },
+        )
+
+        if config.project_dir.exists():
+            if not questionary.confirm(
+                f"Directory {config.project_dir} exists, do you want to continue and update the generated files? "
+                + "This will overwrite your changes in those files."
+            ).ask():
+                logger.warning("Exiting...")
+                exit(0)
+
+        create_new_client(
+            url=url,
+            path=path,
+            config=config,
+        )
+        logger.success("Pipeline created. Learn more at https://dlthub.com/docs. See you next time :)")
+
+    except DltOpenAPITerminalException as exc:
+        logger.error("Encountered terminal exception:")
+        logger.error(exc)
+        logger.info("Exiting...")

dlt_init_openapi/exceptions.py

@@ -1,2 +1,38 @@
 class DltOpenAPIException(Exception):
     pass
+
+
+class DltOpenAPITerminalException(DltOpenAPIException):
+    pass
+
+
+class DltOpenAPINot30Exception(DltOpenAPITerminalException):
+    def __init__(self, swagger_detected: bool = False) -> None:
+
+        swagger_helper = "If this is a Swagger/OpenAPI 2.0 or earlier spec, "
+        if swagger_detected:
+            swagger_helper = "It looks like this is a Swagger/OpenAPI 2.0 spec, "
+
+        convert_helper = (
+            "you can convert it to an openapi 3.0 spec by going to https://editor.swagger.io/, "
+            + "pasting your spec and selecting 'Edit' -> 'Convert to OpenAPI 3.0' from the Menu "
+            + "and then retry with the converted file."
+        )
+
+        super().__init__(
+            "The spec you selected does not appear to be an OpenAPI 3.0 spec. " + swagger_helper + convert_helper
+        )
+
+
+class DltInvalidSpecException(DltOpenAPITerminalException):
+    def __init__(self) -> None:
+
+        super().__init__(
+            "Could not validate selected spec, please provide a valid YAML or JSON OpenAPI 3.0 or higher spec."
+        )
+
+
+class DltUnparseableSpecException(DltOpenAPITerminalException):
+    def __init__(self) -> None:
+
+        super().__init__("Could not parse selected spec, please provide a valid YAML or JSON document.")

dlt_init_openapi/parser/openapi_parser.py

@@ -7,7 +7,7 @@
 from loguru import logger
 from yaml import BaseLoader
 
-from dlt_init_openapi.exceptions import DltOpenAPIException
+from dlt_init_openapi.exceptions import DltInvalidSpecException, DltOpenAPINot30Exception, DltUnparseableSpecException
 from dlt_init_openapi.parser.config import Config
 from dlt_init_openapi.parser.context import OpenapiContext
 from dlt_init_openapi.parser.endpoints import EndpointCollection
@@ -39,30 +39,17 @@ def parse(self, data: bytes) -> None:
         try:
             spec = osp.OpenAPI.parse_obj(self.spec_raw)
         except Exception as e:
-            raise DltOpenAPIException("Could not Validate spec:\n" + str(e)) from e
+            raise DltInvalidSpecException() from e
         logger.success("Spec validation successful")
 
         # check if this is openapi 3.0
-        convert_info = (
-            "you can convert it to an openapi 3.0 spec by going to https://editor.swagger.io/, "
-            + "pasting your spec and selecting 'Edit' -> 'Convert to OpenAPI 3.0' from the Menu "
-            + "and then retry with the converted file."
-        )
         swagger_version = self.spec_raw.get("swagger")
         if swagger_version:
-            logger.error(
-                "The spec you selected appears to be a Swagger/OpenAPI version 2 spec or older, " + convert_info
-            )
-            exit(0)
+            raise DltOpenAPINot30Exception(swagger_detected=True)
 
         openapi_version = self.spec_raw.get("openapi")
         if not openapi_version or not openapi_version.startswith("3"):
-            logger.error(
-                "The spec you selected does not appear to be an OpenAPI 3.0 spec. "
-                + "If this is a a Swagger/OpenAPI version 2 spec or older, "
-                + convert_info
-            )
-            exit(1)
+            raise DltOpenAPINot30Exception(swagger_detected=False)
 
         logger.info("Extracting openapi metadata")
         self.context = OpenapiContext(self.config, spec, self.spec_raw)
@@ -88,19 +75,21 @@ def parse(self, data: bytes) -> None:
 
     def _load_yaml_or_json(self, data: bytes) -> Dict[str, Any]:
         logger.info("Trying to parse spec as JSON")
-
-        data_size = sys.getsizeof(data)
-        if data_size > 1000000:
-            mb = round(data_size / 1000000)
-            logger.warning(f"Spec is around {mb} mb, so parsing might take a while.")
         try:
-            result = json.loads(data.decode())
-            logger.success("Parsed spec as JSON")
+            data_size = sys.getsizeof(data)
+            if data_size > 1000000:
+                mb = round(data_size / 1000000)
+                logger.warning(f"Spec is around {mb} mb, so parsing might take a while.")
+            try:
+                result = json.loads(data.decode())
+                logger.success("Parsed spec as JSON")
+                return result
+            except ValueError:
+                logger.info("No valid JSON found")
+                pass
+            logger.info("Trying to parse spec as YAML")
+            result = yaml.load(data, Loader=BaseLoader)
+            logger.success("Parsed spec as YAML")
             return result
-        except ValueError:
-            logger.info("No valid JSON found")
-            pass
-        logger.info("Trying to parse spec as YAML")
-        result = yaml.load(data, Loader=BaseLoader)
-        logger.success("Parsed spec as YAML")
-        return result
+        except Exception as exc:
+            raise DltUnparseableSpecException() from exc