Added upload manifests endpoint for reachability

Author: dacoburn

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "socketdev"
-version = "3.0.14"
+version = "3.0.15"
 requires-python = ">= 3.9"
 dependencies = [
     'requests',

socketdev/__init__.py

@@ -25,6 +25,7 @@
 from socketdev.analytics import Analytics
 from socketdev.alerttypes import AlertTypes
 from socketdev.basics import Basics
+from socketdev.uploadmanifests import UploadManifests
 from socketdev.log import log
 
 __author__ = "socket.dev"
@@ -74,6 +75,7 @@ def __init__(self, token: str, timeout: int = 1200):
         self.analytics = Analytics(self.api)
         self.alerttypes = AlertTypes(self.api)
         self.basics = Basics(self.api)
+        self.uploadmanifests = UploadManifests(self.api)
 
     @staticmethod
     def set_timeout(timeout: int):

socketdev/uploadmanifests/__init__.py

@@ -0,0 +1,67 @@
+import os
+import logging
+from typing import List, Optional, Union
+from ..utils import Utils
+
+log = logging.getLogger("socketdev")
+
+
+class UploadManifests:
+    def __init__(self, api):
+        self.api = api
+
+    def upload_manifest_files(self, org_slug: str, file_paths: List[str], workspace: Optional[str] = None, base_path: Optional[str] = None, base_paths: Optional[List[str]] = None, use_lazy_loading: bool = True) -> str:
+        """
+        Upload manifest files to Socket API and return tarHash.
+        
+        Args:
+            org_slug: Organization slug
+            file_paths: List of manifest file paths to upload
+            workspace: Base directory path to make paths relative to
+            base_path: Optional base path to strip from key names for cleaner file organization
+            base_paths: Optional list of base paths to strip from key names (takes precedence over base_path)
+            use_lazy_loading: Whether to use lazy file loading (default: True)
+            
+        Returns:
+            str: The tarHash from the upload response
+            
+        Raises:
+            Exception: If upload fails
+        """
+        # Filter to only existing files
+        valid_files = [f for f in file_paths if os.path.exists(f) and os.path.isfile(f)]
+        
+        if not valid_files:
+            raise Exception("No valid manifest files found to upload")
+        
+        # Prepare files for upload using the utility function
+        if use_lazy_loading:
+            loaded_files = Utils.load_files_for_sending_lazy(
+                valid_files, 
+                workspace=workspace, 
+                base_path=base_path, 
+                base_paths=base_paths
+            )
+        else:
+            # Fallback to basic file loading if needed
+            loaded_files = []
+            for file_path in valid_files:
+                key = os.path.basename(file_path)
+                with open(file_path, 'rb') as f:
+                    loaded_files.append((key, (key, f.read())))
+        
+        # Make the upload request
+        path = f"orgs/{org_slug}/upload-manifest-files"
+        response = self.api.do_request(path=path, files=loaded_files, method="POST")
+        
+        if response.status_code != 200:
+            raise Exception(f"Upload failed with status {response.status_code}: {response.text}")
+        
+        result = response.json()
+        tar_hash = result.get('tarHash')
+        
+        if not tar_hash:
+            raise Exception("Server did not return a tarHash")
+            
+        log.info(f"Successfully uploaded {len(valid_files)} manifest files, tarHash: {tar_hash}")
+        return tar_hash

socketdev/utils/__init__.py

@@ -1,4 +1,4 @@
-from typing import Literal, List, Tuple
+from typing import Literal, List, Tuple, Optional
 import logging
 import os
 import weakref
@@ -233,7 +233,7 @@ def validate_integration_type(integration_type: str) -> IntegrationType:
         return integration_type  # type: ignore
 
     @staticmethod
-    def load_files_for_sending_lazy(files: List[str], workspace: str = None, max_open_files: int = 100, base_path: str = None, base_paths: List[str] = None) -> List[Tuple[str, Tuple[str, LazyFileLoader]]]:
+    def load_files_for_sending_lazy(files: List[str], workspace: Optional[str] = None, max_open_files: int = 100, base_path: Optional[str] = None, base_paths: Optional[List[str]] = None) -> List[Tuple[str, Tuple[str, LazyFileLoader]]]:
         """
         Prepares files for sending to the Socket API using lazy loading.
         
@@ -342,4 +342,6 @@ def load_files_for_sending_lazy(files: List[str], workspace: str = None, max_ope
                 lazy_file = LazyFileLoader(file_path, key)
                 payload = (key, (key, lazy_file))
                 send_files.append(payload)
-            return send_files
+
+        log.debug(f"Prepared {len(send_files)} files for lazy loading")
+        return send_files

socketdev/version.py

@@ -1 +1 @@
-__version__ = "3.0.14"
+__version__ = "3.0.15"