Claude's first pass at adding bundle metadata support

Author: nealrichardson

rsconnect/api.py

@@ -9,6 +9,7 @@
 import datetime
 import hashlib
 import hmac
+import json
 import os
 import re
 import sys
@@ -55,7 +56,14 @@
 from .certificates import read_certificate_file
 from .environment import fake_module_file_from_directory
 from .exception import DeploymentFailedException, RSConnectException
-from .http_support import CookieJar, HTTPResponse, HTTPServer, JsonData, append_to_path
+from .http_support import (
+    CookieJar,
+    HTTPResponse,
+    HTTPServer,
+    JsonData,
+    append_to_path,
+    create_multipart_form_data,
+)
 from .log import cls_logged, connect_logger, console_logger, logger
 from .metadata import AppStore, ServerStore
 from .models import (
@@ -76,6 +84,7 @@
 )
 from .snowflake import generate_jwt, get_parameters
 from .timeouts import get_task_timeout, get_task_timeout_help_message
+from .utils_package import compare_semvers
 
 if TYPE_CHECKING:
     import logging
@@ -367,6 +376,26 @@ class RSConnectClientDeployResult(TypedDict):
     title: str | None
 
 
+def server_supports_git_metadata(server_version: Optional[str]) -> bool:
+    """
+    Check if the server version supports git metadata in bundle uploads.
+
+    Git metadata support was added in Connect 2025.11.0.
+
+    :param server_version: The Connect server version string
+    :return: True if the server supports git metadata, False otherwise
+    """
+    if not server_version:
+        return False
+
+    try:
+        return compare_semvers(server_version, "2025.11.0") >= 0
+    except Exception:
+        # If we can't parse the version, assume it doesn't support it
+        logger.debug(f"Unable to parse server version: {server_version}")
+        return False
+
+
 class RSConnectClient(HTTPServer):
     def __init__(self, server: Union[RSConnectServer, SPCSConnectServer], cookies: Optional[CookieJar] = None):
         if cookies is None:
@@ -488,11 +517,34 @@ def content_create(self, name: str) -> ContentItemV1:
         response = self._server.handle_bad_response(response)
         return response
 
-    def content_upload_bundle(self, content_guid: str, tarball: typing.IO[bytes]) -> BundleMetadata:
-        response = cast(
-            Union[BundleMetadata, HTTPResponse], self.post("v1/content/%s/bundles" % content_guid, body=tarball)
-        )
-        response = self._server.handle_bad_response(response)
+    def content_upload_bundle(
+        self, content_guid: str, tarball: typing.IO[bytes], metadata: Optional[dict[str, str]] = None
+    ) -> BundleMetadata:
+        """
+        Upload a bundle to the server.
+
+        :param app_id: Application ID
+        :param tarball: Bundle tarball file object
+        :param metadata: Optional metadata dictionary (e.g., git metadata)
+        :return: ContentItemV0 with bundle information
+        """
+        if metadata:
+            # Use multipart form upload when metadata is provided
+            tarball_content = tarball.read()
+            fields = {
+                "archive": ("bundle.tar.gz", tarball_content, "application/x-tar"),
+                "metadata": json.dumps(metadata),
+            }
+            body, content_type = create_multipart_form_data(fields)
+            response = cast(
+                Union[BundleMetadata, HTTPResponse],
+                self.post("v1/content/%s/bundles" % content_guid, body=body, headers={"Content-Type": content_type}),
+            )
+        else:
+            response = cast(
+                Union[BundleMetadata, HTTPResponse], self.post("v1/content/%s/bundles" % content_guid, body=tarball)
+            )
+            response = self._server.handle_bad_response(response)
         return response
 
     def content_update(self, content_guid: str, updates: Mapping[str, str | None]) -> ContentItemV1:
@@ -571,6 +623,7 @@ def deploy(
         tarball: IO[bytes],
         env_vars: Optional[dict[str, str]] = None,
         activate: bool = True,
+        metadata: Optional[dict[str, str]] = None,
     ) -> RSConnectClientDeployResult:
         if app_id is None:
             if app_name is None:
@@ -598,7 +651,7 @@ def deploy(
             result = self._server.handle_bad_response(result)
             app["title"] = app_title
 
-        app_bundle = self.content_upload_bundle(app_guid, tarball)
+        app_bundle = self.content_upload_bundle(app_guid, tarball, metadata=metadata)
 
         task = self.content_deploy(app_guid, app_bundle["id"], activate=activate)
 
@@ -724,6 +777,7 @@ def __init__(
         visibility: Optional[str] = None,
         disable_env_management: Optional[bool] = None,
         env_vars: Optional[dict[str, str]] = None,
+        metadata: Optional[dict[str, str]] = None,
     ) -> None:
         self.remote_server: TargetableServer
         self.client: RSConnectClient | PositClient
@@ -737,6 +791,7 @@ def __init__(
         self.visibility = visibility
         self.disable_env_management = disable_env_management
         self.env_vars = env_vars
+        self.metadata = metadata
         self.app_mode: AppMode | None = None
         self.app_store: AppStore = AppStore(fake_module_file_from_directory(self.path))
         self.app_store_version: int | None = None
@@ -785,6 +840,7 @@ def fromConnectServer(
         visibility: Optional[str] = None,
         disable_env_management: Optional[bool] = None,
         env_vars: Optional[dict[str, str]] = None,
+        metadata: Optional[dict[str, str]] = None,
     ):
         return cls(
             ctx=ctx,
@@ -807,6 +863,7 @@ def fromConnectServer(
             visibility=visibility,
             disable_env_management=disable_env_management,
             env_vars=env_vars,
+            metadata=metadata,
         )
 
     def output_overlap_header(self, previous: bool) -> bool:
@@ -1069,6 +1126,7 @@ def deploy_bundle(self, activate: bool = True):
                 self.bundle,
                 self.env_vars,
                 activate=activate,
+                metadata=self.metadata,
             )
             self.deployed_info = result
             return self

rsconnect/git_metadata.py

@@ -0,0 +1,180 @@
+"""
+Git metadata detection utilities for bundle uploads
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from os.path import abspath, dirname, exists, join
+from typing import Optional
+from urllib.parse import urlparse
+
+from .log import logger
+
+
+def _run_git_command(args: list[str], cwd: str) -> Optional[str]:
+    """
+    Run a git command and return its output.
+
+    :param args: git command arguments
+    :param cwd: working directory
+    :return: command output or None if command failed
+    """
+    try:
+        result = subprocess.run(
+            ["git"] + args,
+            cwd=cwd,
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+        if result.returncode == 0:
+            return result.stdout.strip()
+        return None
+    except (subprocess.SubprocessError, FileNotFoundError, OSError):
+        return None
+
+
+def is_git_repo(directory: str) -> bool:
+    """
+    Check if directory is inside a git repository.
+
+    :param directory: directory to check
+    :return: True if inside a git repo, False otherwise
+    """
+    result = _run_git_command(["rev-parse", "--git-dir"], directory)
+    return result is not None
+
+
+def has_uncommitted_changes(directory: str) -> bool:
+    """
+    Check if the git repository has uncommitted changes.
+
+    :param directory: directory to check
+    :return: True if there are uncommitted changes
+    """
+    # Check for staged and unstaged changes
+    result = _run_git_command(["status", "--porcelain"], directory)
+    return bool(result)
+
+
+def get_git_commit(directory: str) -> Optional[str]:
+    """
+    Get the current git commit SHA.
+
+    :param directory: directory to check
+    :return: commit SHA or None
+    """
+    return _run_git_command(["rev-parse", "HEAD"], directory)
+
+
+def get_git_branch(directory: str) -> Optional[str]:
+    """
+    Get the current git branch name or tag.
+
+    :param directory: directory to check
+    :return: branch/tag name or None
+    """
+    # First try to get branch name
+    branch = _run_git_command(["rev-parse", "--abbrev-ref", "HEAD"], directory)
+
+    # If we're in detached HEAD state, try to get tag
+    if branch == "HEAD":
+        tag = _run_git_command(["describe", "--exact-match", "--tags"], directory)
+        if tag:
+            return tag
+
+    return branch
+
+
+def get_git_remote_url(directory: str, remote: str = "origin") -> Optional[str]:
+    """
+    Get the URL of a git remote.
+
+    :param directory: directory to check
+    :param remote: remote name (default: "origin")
+    :return: remote URL or None
+    """
+    return _run_git_command(["remote", "get-url", remote], directory)
+
+
+def normalize_git_url_to_https(url: Optional[str]) -> Optional[str]:
+    """
+    Normalize a git URL to HTTPS format.
+
+    Converts SSH URLs like git@github.com:user/repo.git to
+    https://github.com/user/repo.git
+
+    :param url: git URL to normalize
+    :return: normalized HTTPS URL or original if already HTTPS/not recognized
+    """
+    if not url:
+        return url
+
+    # Already HTTPS
+    if url.startswith("https://"):
+        return url
+
+    # Handle git@ SSH format
+    if url.startswith("git@"):
+        # git@github.com:user/repo.git -> https://github.com/user/repo.git
+        # Remove git@ prefix
+        url = url[4:]
+        # Replace first : with /
+        url = url.replace(":", "/", 1)
+        # Add https://
+        return f"https://{url}"
+
+    # Handle ssh:// format
+    if url.startswith("ssh://"):
+        # ssh://git@github.com/user/repo.git -> https://github.com/user/repo.git
+        parsed = urlparse(url)
+        if parsed.hostname:
+            path = parsed.path
+            return f"https://{parsed.hostname}{path}"
+
+    # Return as-is if we can't normalize
+    return url
+
+
+def detect_git_metadata(directory: str, remote: str = "origin") -> dict[str, str]:
+    """
+    Detect git metadata for the given directory.
+
+    :param directory: directory to inspect
+    :param remote: git remote name to use (default: "origin")
+    :return: dictionary with source, source_repo, source_branch, source_commit keys
+    """
+    metadata: dict[str, str] = {}
+
+    if not is_git_repo(directory):
+        logger.debug(f"Directory {directory} is not a git repository")
+        return metadata
+
+    # Get commit SHA
+    commit = get_git_commit(directory)
+    if commit:
+        # Check for uncommitted changes
+        if has_uncommitted_changes(directory):
+            commit = f"{commit}-dirty"
+        metadata["source_commit"] = commit
+
+    # Get branch/tag
+    branch = get_git_branch(directory)
+    if branch:
+        metadata["source_branch"] = branch
+
+    # Get remote URL and normalize to HTTPS
+    remote_url = get_git_remote_url(directory, remote)
+    if remote_url:
+        normalized_url = normalize_git_url_to_https(remote_url)
+        if normalized_url:
+            metadata["source_repo"] = normalized_url
+
+    # Always set source to "git" if we got any metadata
+    if metadata:
+        metadata["source"] = "git"
+        logger.debug(f"Detected git metadata: {metadata}")
+
+    return metadata