Remove unnecessary methods from swagger processing; Fix path errors.

Author: schloerke

extensions/sdk-assistant/_prompt.xml

@@ -1,5 +1,5 @@
 This file is a merged representation of the entire codebase, combining all repository files into a single document.
-Generated by Repomix on: 2025-01-22T22:04:54.185Z
+Generated by Repomix on: 2025-01-28T17:21:45.629Z
 
 <file_summary>
 This section contains a summary of this file.
@@ -918,6 +918,15 @@ def update_dict_values(obj: dict[str, Any], /, **kwargs: Any) -> None:
     * https://github.com/posit-dev/posit-sdk-py/pull/366#discussion_r1887845267
     """
     ...
+
+def is_local() -> bool:
+    """Returns true if called from a piece of content running on a Connect server.
+
+    The connect server will always set the environment variable `RSTUDIO_PRODUCT=CONNECT`.
+    We can use this environment variable to determine if the content is running locally
+    or on a Connect server.
+    """
+    ...
 </file>
 
 <file path="connect/auth.pyi">
@@ -1252,12 +1261,14 @@ class Client(ContextManager):
         """
         ...
 
-    @requires("2025.01.0-dev")
+    @requires("2025.01.0")
     def with_user_session_token(self, token: str) -> Client:
         """Create a new Client scoped to the user specified in the user session token.
 
         Create a new Client instance from a user session token exchange for an api key scoped to the
-        user specified in the token.
+        user specified in the token (the user viewing your app). If running your application locally,
+        a user session token will not exist, which will cause this method to result in an error needing
+        to be handled in your application.
 
         Parameters
         ----------

extensions/sdk-assistant/uv_update_prompt.py

@@ -33,7 +33,10 @@ def cleanup() -> None:
         path = here / f
         if path.exists():
             print("Removing path:", path.relative_to(here))
-            shutil.rmtree(path)
+            if path.is_file():
+                path.unlink()
+            else:
+                shutil.rmtree(path)
     print("--\n")
 
 

extensions/sdk-assistant/uv_update_swagger.py

@@ -6,7 +6,6 @@
 # ///
 from __future__ import annotations
 
-from copy import deepcopy
 import pathlib
 import os
 import json
@@ -18,177 +17,23 @@
 T = TypeVar("T")
 
 
-def find_value(root, path):
-    """
-    Find a value in an object graph.
-
-    This function is used to follow the specified path through the object graph at root
-    and return the item in the graph, if any, that the path refers to.
-
-    :param root: the root of the object graph to traverse.
-    :param path: the path through the graph to take.
-    :return: the resulting value or None.
-    """
-    if isinstance(path, str):
-        path = path.split("/")
-    parent = root
-    for part in path:
-        if part in parent:
-            parent = parent[part]
-        else:
-            return None
-    return parent
-
-
-def expand_refs(
-    document, obj
-) -> Any:  # Use `Any` for return type to hack around typing requirement
-    """
-    Expands `ref`s in the given object.
-
-    Returns an object semantically equivalent to the original but with references expanded.
-
-    Parameters
-    ----------
-    document
-        the master swagger document containing the responses and definitions.
-    obj
-        is either a normal swagger object, a ref object, or a swagger object with a schema.
-    """
-    if isinstance(obj, list):
-        return [expand_refs(document, item) for item in obj]
-    elif isinstance(obj, dict):
-        if "$ref" in obj:
-            ref_path = obj["$ref"].strip("#/").split("/")
-            ref_value = find_value(document, ref_path)
-            if ref_value is None:
-                raise RuntimeError(
-                    f"Reference {obj['$ref']} not found in the document."
-                )
-            return expand_refs(document, ref_value)
-        else:
-            return {key: expand_refs(document, value) for key, value in obj.items()}
-    else:
-        return obj
-
-
 class SwaggerOperation(TypedDict, total=False):
-    operationId: str
-    tags: list[str]
     summary: str
-    description: str
-    parameters: list[dict[str, Any]]
-    responses: dict[str, Any]
 
 
 class SwaggerDocument(TypedDict):
-    paths: NotRequired[dict[str, SwaggerOperation]]
-    parameters: NotRequired[dict[str, Any]]
-    responses: NotRequired[dict[str, Any]]
-    definitions: NotRequired[dict[str, Any]]
-
-
-def expand_all_references(document: SwaggerDocument) -> SwaggerDocument:
-    """
-    Expands all JSON references.
-
-    Expands all references ($ref) in the merged swagger document by replacing them with
-    their full definitions.
-
-    This returns a new document with all references expanded.
-
-    Arguments
-    ---------
-    document
-        The dictionary representing the Swagger document to process
-
-    Returns
-    -------
-    :
-        The processed Swagger document with all references expanded.
-    """
-    ret_document = deepcopy(document)
-    # List of error response keys to ignore
-    error_responses = [
-        "BadRequest",
-        "Unauthorized",
-        "PaymentRequired",
-        "Forbidden",
-        "NotFound",
-        "Conflict",
-        "APIError",
-        "InternalServerError",
-    ]
-
-    # We need to expand refs in paths
-    if "paths" in ret_document:
-        for _path, operations in ret_document["paths"].items():
-            for _method, operation in operations.items():
-                if not isinstance(operation, dict):
-                    continue
-                # Expand refs in parameters
-                if "parameters" in operation:
-                    operation["parameters"] = expand_refs(
-                        ret_document, operation["parameters"]
-                    )
-
-                # Expand refs in responses
-                if "responses" in operation:
-                    for code, response in operation["responses"].items():
-                        if "schema" in response and code not in error_responses:
-                            response["schema"] = expand_refs(
-                                ret_document, response["schema"]
-                            )
-
-    # Expand refs in top-level parameters
-    if "parameters" in ret_document:
-        ret_document["parameters"] = expand_refs(
-            ret_document, ret_document["parameters"]
-        )
-
-    # Expand refs in top-level responses, ignoring error responses
-    if "responses" in ret_document:
-        for response_key, response_value in ret_document["responses"].items():
-            if response_key not in error_responses:
-                ret_document["responses"][response_key] = expand_refs(
-                    ret_document, response_value
-                )
-
-    # Expand refs in definitions
-    if "definitions" in ret_document:
-        ret_document["definitions"] = expand_refs(
-            ret_document, ret_document["definitions"]
-        )
-
-    return ret_document
-
-
-def require_swagger():
-    if not (here / "_swagger.json").exists():
-        import urllib.request
-
-        urllib.request.urlretrieve(
-            "https://docs.posit.co/connect/api/swagger.json",
-            here / "_swagger.json",
-        )
-
-    doc = json.load(here / "_swagger.json")
-
-    swagger = expand_all_references(doc)
-    return swagger
+    paths: dict[str, SwaggerOperation]
 
 
 class OperationDef(TypedDict):
-    name: str
-    tags: list[str]
     method: str
     route: str
-    definition: dict[str, Any]
+    summary: str
 
 
 def transform_swagger_to_operation_dict(
     swagger_dict: SwaggerDocument,
-) -> dict[str, OperationDef]:
+) -> list[OperationDef]:
     """
     Swagger to operation dictionary transformation.
 
@@ -204,49 +49,78 @@ def transform_swagger_to_operation_dict(
     :
         A dictionary where each key is an operation ID and the value is the operation definition.
     """
-    operation_dict = {}
+    operation_list: list[OperationDef] = []
+
+    if "paths" not in swagger_dict:
+        raise ValueError(
+            "The Swagger document `swagger_dict=` does not contain a 'paths' key."
+        )
 
     if "paths" in swagger_dict:
         for route, operations in swagger_dict["paths"].items():
+            if not isinstance(route, str):
+                raise ValueError(
+                    f"Expected route to be a string, but got {type(route)}."
+                )
             for method, operation in operations.items():
+                if not isinstance(method, str):
+                    raise ValueError(
+                        f"Expected method to be a string, but got {type(method)}."
+                    )
                 if not isinstance(operation, dict):
-                    continue
-                if "operationId" in operation:
-                    operation_id = operation["operationId"]
-                    tags = operation["tags"] if "tags" in operation else []
-                    # response_parent = operation["responses"]["200"] or operation["responses"]["204"]
-                    # if response_parent and "response"
-                    operation_dict[operation_id] = {
-                        "name": operation_id,
-                        "tags": tags,
+                    raise ValueError(
+                        f"Expected operation to be a dictionary, but got {type(operation)}."
+                    )
+                if "summary" not in operation:
+                    raise ValueError(
+                        f"Expected operation to have a 'summary' key, but got {operation}."
+                    )
+                summary = operation["summary"]
+                if not isinstance(summary, str):
+                    raise ValueError(
+                        f"Expected summary to be a string, but got {type(summary)}."
+                    )
+                operation_list.append(
+                    {
                         "method": method,
                         "route": route,
-                        "definition": operation,
+                        "summary": summary,
                     }
+                )
 
-    return operation_dict
+    return operation_list
 
 
 def main():
-    swagger = require_swagger()
+    if not (here / "_swagger.json").exists():
+        import urllib.request
+
+        urllib.request.urlretrieve(
+            "https://docs.posit.co/connect/api/swagger.json",
+            here / "_swagger.json",
+        )
+
+    swagger = json.loads((here / "_swagger.json").read_text())
 
     operations = transform_swagger_to_operation_dict(swagger)
 
+    # Write out the swagger portion of the instructions with a preamble and a
+    # list of all the API routes and their short summaries.
     with open(here / "_swagger_prompt.md", "w") as f:
         f.write(
             "If an answer can not be resolved, suggest to the user that they can explore calling these API routes themselves. Never produce code that calls these routes as we do not know the return type or successful status codes.\n\nAPI Routes:\n"
             ""
         )
 
-        for operation in operations.values():
-            # `"GET /v1/tasks/{id} Get task details"`
+        for operation in operations:
+            # `"* GET /v1/tasks/{id} Get task details"`
             f.write(
                 "* "
                 + operation["method"].upper()
                 + " "
                 + operation["route"]
                 + " "
-                + operation["definition"]["summary"].replace("\n", " ").strip()
+                + operation["summary"].replace("\n", " ").strip()
                 + "\n",
             )