new: Changes for CLI/TechDocs spec compatibility (#624)

Co-authored-by: Zhiwei Liang <[email protected]>

Author: lgarber-akamai

.gitignore

@@ -13,3 +13,4 @@ test/.env
 .tmp*
 MANIFEST
 venv
+openapi*.yaml

linodecli/__init__.py

@@ -51,7 +51,11 @@
     or TEST_MODE
 )
 
-cli = CLI(VERSION, handle_url_overrides(BASE_URL), skip_config=skip_config)
+cli = CLI(
+    VERSION,
+    handle_url_overrides(BASE_URL, override_path=True),
+    skip_config=skip_config,
+)
 
 
 def main():  # pylint: disable=too-many-branches,too-many-statements

linodecli/api_request.py

@@ -13,7 +13,7 @@
 from requests import Response
 
 from linodecli.exit_codes import ExitCodes
-from linodecli.helpers import API_CA_PATH
+from linodecli.helpers import API_CA_PATH, API_VERSION_OVERRIDE
 
 from .baked.operation import (
     ExplicitEmptyListValue,
@@ -185,14 +185,22 @@ def _build_filter_header(
 
 
 def _build_request_url(ctx, operation, parsed_args) -> str:
-    target_server = handle_url_overrides(
+    url_base = handle_url_overrides(
         operation.url_base,
         host=ctx.config.get_value("api_host"),
-        version=ctx.config.get_value("api_version"),
         scheme=ctx.config.get_value("api_scheme"),
     )
 
-    result = f"{target_server}{operation.url_path}".format(**vars(parsed_args))
+    result = f"{url_base}{operation.url_path}".format(
+        # {apiVersion} is defined in the endpoint paths for
+        # the TechDocs API specs
+        apiVersion=(
+            API_VERSION_OVERRIDE
+            or ctx.config.get_value("api_version")
+            or operation.default_api_version
+        ),
+        **vars(parsed_args),
+    )
 
     if operation.method == "get":
         result += f"?page={ctx.page}&page_size={ctx.page_size}"

linodecli/baked/operation.py

@@ -11,10 +11,11 @@
 from collections import defaultdict
 from getpass import getpass
 from os import environ, path
-from typing import Any, Dict, List, Tuple
+from typing import Any, Dict, List, Optional, Tuple
+from urllib.parse import urlparse
 
 import openapi3.paths
-from openapi3.paths import Operation
+from openapi3.paths import Operation, Parameter
 
 from linodecli.baked.request import OpenAPIFilteringRequest, OpenAPIRequest
 from linodecli.baked.response import OpenAPIResponse
@@ -295,7 +296,9 @@ class OpenAPIOperation:
     This is the class that should be pickled when building the CLI.
     """
 
-    def __init__(self, command, operation: Operation, method, params):
+    def __init__(
+        self, command, operation: Operation, method, params
+    ):  # pylint: disable=too-many-locals,too-many-branches,too-many-statements
         """
         Wraps an openapi3.Operation object and handles pulling out values relevant
         to the Linode CLI.
@@ -309,16 +312,25 @@ def __init__(self, command, operation: Operation, method, params):
         self.response_model = None
         self.allowed_defaults = None
 
+        # The legacy spec uses "200" (str) in response keys
+        # while the new spec uses 200 (int).
+        response_key = "200" if "200" in operation.responses else 200
+
         if (
-            "200" in operation.responses
-            and "application/json" in operation.responses["200"].content
+            response_key in operation.responses
+            and "application/json" in operation.responses[response_key].content
         ):
             self.response_model = OpenAPIResponse(
-                operation.responses["200"].content["application/json"]
+                operation.responses[response_key].content["application/json"]
             )
 
         if method in ("post", "put") and operation.requestBody:
-            if "application/json" in operation.requestBody.content:
+            content = operation.requestBody.content
+
+            if (
+                "application/json" in content
+                and content["application/json"].schema is not None
+            ):
                 self.request = OpenAPIRequest(
                     operation.requestBody.content["application/json"]
                 )
@@ -346,30 +358,27 @@ def __init__(self, command, operation: Operation, method, params):
 
         self.summary = operation.summary
         self.description = operation.description.split(".")[0]
-        self.params = [OpenAPIOperationParameter(c) for c in params]
 
-        # These fields must be stored separately
-        # to allow them to be easily modified
-        # at runtime.
-        self.url_base = (
-            operation.servers[0].url
-            if operation.servers
-            else operation._root.servers[0].url
-        )
+        # The apiVersion attribute should not be specified as a positional argument
+        self.params = [
+            OpenAPIOperationParameter(param)
+            for param in params
+            if param.name not in {"apiVersion"}
+        ]
 
-        self.url_path = operation.path[-2]
+        self.url_base, self.url_path, self.default_api_version = (
+            self._get_api_url_components(operation, params)
+        )
 
         self.url = self.url_base + self.url_path
 
-        docs_url = None
-        tags = operation.tags
-        if tags is not None and len(tags) > 0 and len(operation.summary) > 0:
-            tag_path = self._flatten_url_path(tags[0])
-            summary_path = self._flatten_url_path(operation.summary)
-            docs_url = (
-                f"https://www.linode.com/docs/api/{tag_path}/#{summary_path}"
+        self.docs_url = self._resolve_operation_docs_url(operation)
+
+        if self.docs_url is None:
+            print(
+                f"INFO: Could not resolve docs URL for {operation}",
+                file=sys.stderr,
             )
-        self.docs_url = docs_url
 
         code_samples_ext = operation.extensions.get("code-samples")
         self.samples = (
@@ -401,6 +410,85 @@ def _flatten_url_path(tag: str) -> str:
         new_tag = re.sub(r"[^a-z ]", "", new_tag).replace(" ", "-")
         return new_tag
 
+    @staticmethod
+    def _resolve_api_version(
+        params: List[Parameter], server_url: str
+    ) -> Optional[str]:
+        """
+        Returns the API version for a given list of params and target URL.
+
+        :param params: The params for this operation's endpoint path.
+        :type params: List[Parameter]
+        :param server_url: The URL of server for this operation.
+        :type server_url: str
+
+        :returns: The default API version if the URL has a version, else None.
+        :rtype: Optional[str]
+        """
+
+        # Remove empty segments from the URL path, stripping the first,
+        # last and any duplicate slashes if necessary.
+        # There shouldn't be a case where this is needed, but it's
+        # always good to make things more resilient :)
+        url_path_segments = [
+            seg for seg in urlparse(server_url).path.split("/") if len(seg) > 0
+        ]
+        if len(url_path_segments) > 0:
+            return "/".join(url_path_segments)
+
+        version_param = next(
+            (
+                param
+                for param in params
+                if param.name == "apiVersion" and param.in_ == "path"
+            ),
+            None,
+        )
+        if version_param is not None:
+            return version_param.schema.default
+
+        return None
+
+    @staticmethod
+    def _get_api_url_components(
+        operation: Operation, params: List[Parameter]
+    ) -> Tuple[str, str, str]:
+        """
+        Returns the URL components for a given operation.
+
+        :param operation: The operation to get the URL components for.
+        :type operation: Operation
+        :param params: The parameters for this operation's route.
+        :type params: List[Parameter]
+
+        :returns: The base URL, path, and default API version of the operation.
+        :rtype: Tuple[str, str, str]
+        """
+
+        url_server = (
+            operation.servers[0].url
+            if operation.servers
+            # pylint: disable-next=protected-access
+            else operation._root.servers[0].url
+        )
+
+        url_base = urlparse(url_server)._replace(path="").geturl()
+        url_path = operation.path[-2]
+
+        api_version = OpenAPIOperation._resolve_api_version(params, url_server)
+        if api_version is None:
+            raise ValueError(
+                f"Failed to resolve API version for operation {operation}"
+            )
+
+        # The apiVersion is only specified in the new-style OpenAPI spec,
+        # so we need to manually insert it into the path to maintain
+        # backwards compatibility
+        if "{apiVersion}" not in url_path:
+            url_path = "/{apiVersion}" + url_path
+
+        return url_base, url_path, api_version
+
     def process_response_json(
         self, json: Dict[str, Any], handler: OutputHandler
     ):  # pylint: disable=redefined-outer-name
@@ -437,6 +525,7 @@ def _add_args_filter(self, parser: argparse.ArgumentParser):
 
         # build args for filtering
         filterable_args = []
+
         for attr in self.response_model.attrs:
             if not attr.filterable:
                 continue
@@ -715,3 +804,45 @@ def parse_args(self, args: Any) -> argparse.Namespace:
             self._validate_parent_child_conflicts(parsed)
 
         return self._handle_list_items(list_items, parsed)
+
+    @staticmethod
+    def _resolve_operation_docs_url_legacy(
+        operation: Operation,
+    ) -> Optional[str]:
+        """
+        Gets the docs URL for a given operation in the legacy OpenAPI spec.
+
+        :param operation: The target openapi3.Operation to get the docs URL for.
+        :type operation: str
+
+        :returns: The docs URL if it can be resolved, else None
+        :rtype: Optional[str]
+        """
+        tags = operation.tags
+        if tags is None or len(tags) < 1 or len(operation.summary) < 1:
+            return None
+
+        tag_path = OpenAPIOperation._flatten_url_path(tags[0])
+        summary_path = OpenAPIOperation._flatten_url_path(operation.summary)
+        return f"https://www.linode.com/docs/api/{tag_path}/#{summary_path}"
+
+    @staticmethod
+    def _resolve_operation_docs_url(operation: Operation) -> Optional[str]:
+        """
+        Gets the docs URL for a given OpenAPI operation.
+
+        :param operation: The target openapi3.Operation to get the docs URL for.
+        :type operation: str
+
+        :returns: The docs URL if it can be resolved, else None
+        :rtype: Optional[str]
+        """
+        # Case for TechDocs
+        if (
+            operation.externalDocs is not None
+            and operation.externalDocs.url is not None
+        ):
+            return operation.externalDocs.url
+
+        # Case for legacy docs
+        return OpenAPIOperation._resolve_operation_docs_url_legacy(operation)