add commands to list and fetch jumpstart examples

Author: edavidaja

rsconnect/actions_content.py

@@ -23,6 +23,7 @@
     ContentItemV1,
     VersionSearchFilter,
 )
+from .utils_package import compare_semvers
 
 _content_build_store: ContentBuildStore | None = None
 
@@ -503,3 +504,23 @@ def _order_content_results(
         result = sorted(result, key=lambda c: c["created_time"], reverse=True)
 
     return list(result)
+
+
+def list_examples(connect_server: RSConnectServer):
+    with RSConnectClient(connect_server) as client:
+        connect_version = client.server_settings()["version"]
+        has_public_examples = compare_semvers(connect_version, "2024.05.0")
+        result = client.examples_list() if has_public_examples in [0, 1] else client.examples_list_legacy()
+    return result
+
+
+def download_example(connect_server: RSConnectServer, example_name: str):
+    with RSConnectClient(connect_server) as client:
+        connect_version = client.server_settings()["version"]
+        has_public_examples = compare_semvers(connect_version, "2024.05.0")
+        result = (
+            client.examples_download(example_name)
+            if has_public_examples in [0, 1]
+            else client.examples_download_legacy(example_name)
+        )
+    return result

rsconnect/api.py

@@ -68,6 +68,7 @@
     ContentItemV1,
     DeleteInputDTO,
     DeleteOutputDTO,
+    Examples,
     ListEntryOutputDTO,
     PyInfo,
     ServerSettings,
@@ -389,6 +390,28 @@ def content_build(self, content_guid: str, bundle_id: Optional[str] = None) -> B
         response = self._server.handle_bad_response(response)
         return response
 
+    def examples_list(self) -> list[Examples]:
+        response = cast(Union[List[Examples], HTTPResponse], self.get("v1/examples"))
+        response = self._server.handle_bad_response(response)
+        return response
+
+    # todo: delete me in October of 2025
+    def examples_list_legacy(self) -> list[Examples]:
+        response = cast(Union[List[Examples], HTTPResponse], self.get("v1/experimental/examples"))
+        response = self._server.handle_bad_response(response)
+        return response
+
+    def examples_download(self, example_name: str) -> HTTPResponse:
+        response = cast(HTTPResponse, self.get("v1/examples/%s/zip" % example_name, decode_response=False))
+        response = self._server.handle_bad_response(response, is_httpresponse=True)
+        return response
+
+    # todo: delete me in October of 2025
+    def examples_download_legacy(self, example_name: str) -> HTTPResponse:
+        response = cast(HTTPResponse, self.get("v1/experimental/examples/%s/zip" % example_name, decode_response=False))
+        response = self._server.handle_bad_response(response, is_httpresponse=True)
+        return response
+
     def system_caches_runtime_list(self) -> list[ListEntryOutputDTO]:
         response = cast(Union[List[ListEntryOutputDTO], HTTPResponse], self.get("v1/system/caches/runtime"))
         response = self._server.handle_bad_response(response)

rsconnect/main.py

@@ -44,8 +44,10 @@
     build_remove_content,
     build_start,
     download_bundle,
+    download_example,
     emit_build_log,
     get_content,
+    list_examples,
     search_content,
 )
 from .api import RSConnectClient, RSConnectExecutor, RSConnectServer
@@ -60,8 +62,8 @@
     make_manifest_bundle,
     make_notebook_html_bundle,
     make_notebook_source_bundle,
-    make_voila_bundle,
     make_tensorflow_bundle,
+    make_voila_bundle,
     read_manifest_app_mode,
     validate_entry_point,
     validate_extra_files,
@@ -2815,6 +2817,88 @@ def system_caches_delete(
         ce.delete_runtime_cache(language, version, image_name, dry_run)
 
 
+@cli.group(no_args_is_help=True, help="Fetch Posit Connect jumpstart examples.")
+def examples():
+    pass
+
+
+@examples.command(
+    name="list",
+    short_help="List jumpstart examples on a Posit Connect server.",
+)
+@server_args
+@click.pass_context
+def examples_list(
+    ctx: click.Context,
+    name: str,
+    server: Optional[str],
+    api_key: Optional[str],
+    insecure: bool,
+    cacert: Optional[str],
+    verbose: int,
+):
+    set_verbosity(verbose)
+    output_params(ctx, locals().items())
+    with cli_feedback("", stderr=True):
+        ce = RSConnectExecutor(ctx, name, server, api_key, insecure, cacert, logger=None).validate_server()
+        if not isinstance(ce.remote_server, RSConnectServer):
+            raise RSConnectException("rsconnect examples list` requires a Posit Connect server.")
+        examples = list_examples(ce.remote_server)
+        result = [{"name": ex["name"], "description": ex["description"]} for ex in examples]
+        json.dump(result, sys.stdout, indent=2)
+
+
+@examples.command(
+    name="download",
+    short_help="Download a jumpstart example from a Posit Connect server.",
+)
+@server_args
+@click.option(
+    "--example",
+    required=True,
+    help="The name of the example to download.",
+)
+@click.option(
+    "--output",
+    "-o",
+    type=click.Path(),
+    required=True,
+    help="Defines the output location for the download.",
+)
+@click.option(
+    "--overwrite",
+    is_flag=True,
+    help="Overwrite the output file if it already exists.",
+)
+@click.pass_context
+def examples_download(
+    ctx: click.Context,
+    name: Optional[str],
+    server: Optional[str],
+    api_key: Optional[str],
+    insecure: bool,
+    cacert: Optional[str],
+    example: str,
+    output: str,
+    overwrite: bool,
+    verbose: int,
+):
+    set_verbosity(verbose)
+    output_params(ctx, locals().items())
+    with cli_feedback("", stderr=True):
+        ce = RSConnectExecutor(ctx, name, server, api_key, insecure, cacert, logger=None).validate_server()
+        if not isinstance(ce.remote_server, RSConnectServer):
+            raise RSConnectException("`rsconnect examples download` requires a Posit Connect server.")
+        if exists(output) and not overwrite:
+            raise RSConnectException("The output file already exists: %s" % output)
+
+        result = download_example(ce.remote_server, example)
+        if not isinstance(result.response_body, bytes):
+            raise RSConnectException("The response body must be bytes (not string or None).")
+        with open(output, "wb") as f:
+            f.write(result.response_body)
+
+
 if __name__ == "__main__":
     cli()
     click.echo()

rsconnect/models.py

@@ -617,3 +617,13 @@ class UserRecord(TypedDict):
     locked: bool
     guid: str
     preferences: dict[str, object]
+
+
+class Examples(TypedDict):
+    name: str
+    type: str
+    title: str
+    description: str
+    files: list[str]
+    requirements: list[str]
+    links: list[dict[str, str]]