add _load_instance_from_remote_storage

litellm/proxy/common_utils/load_config_utils.py

@@ -1,3 +1,6 @@
+import os
+import tempfile
+
 import yaml
 
 from litellm._logging import verbose_proxy_logger
@@ -62,6 +65,98 @@ async def get_config_file_contents_from_gcs(bucket_name, object_key):
         return None
 
 
+def download_python_file_from_s3(bucket_name, object_key, local_file_path):
+    """
+    Download a Python file from S3 and save it to local filesystem.
+    
+    Args:
+        bucket_name (str): S3 bucket name
+        object_key (str): S3 object key (file path in bucket)
+        local_file_path (str): Local path where file should be saved
+        
+    Returns:
+        bool: True if successful, False otherwise
+    """
+    try:
+        import boto3
+        from botocore.credentials import Credentials
+
+        from litellm.main import bedrock_converse_chat_completion
+
+        credentials: Credentials = bedrock_converse_chat_completion.get_credentials()
+        s3_client = boto3.client(
+            "s3",
+            aws_access_key_id=credentials.access_key,
+            aws_secret_access_key=credentials.secret_key,
+            aws_session_token=credentials.token,
+        )
+        
+        verbose_proxy_logger.debug(
+            f"Downloading Python file {object_key} from S3 bucket: {bucket_name}"
+        )
+        response = s3_client.get_object(Bucket=bucket_name, Key=object_key)
+        
+        # Read the file contents
+        file_contents = response["Body"].read().decode("utf-8")
+        
+        # Ensure directory exists
+        os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
+        
+        # Write to local file
+        with open(local_file_path, 'w') as f:
+            f.write(file_contents)
+            
+        verbose_proxy_logger.debug(f"Python file downloaded successfully to {local_file_path}")
+        return True
+
+    except ImportError as e:
+        verbose_proxy_logger.error(f"ImportError: {str(e)}")
+        return False
+    except Exception as e:
+        verbose_proxy_logger.error(f"Error downloading Python file: {str(e)}")
+        return False
+
+
+async def download_python_file_from_gcs(bucket_name, object_key, local_file_path):
+    """
+    Download a Python file from GCS and save it to local filesystem.
+    
+    Args:
+        bucket_name (str): GCS bucket name
+        object_key (str): GCS object key (file path in bucket)
+        local_file_path (str): Local path where file should be saved
+        
+    Returns:
+        bool: True if successful, False otherwise
+    """
+    try:
+        from litellm.integrations.gcs_bucket.gcs_bucket import GCSBucketLogger
+
+        gcs_bucket = GCSBucketLogger(
+            bucket_name=bucket_name,
+        )
+        file_contents = await gcs_bucket.download_gcs_object(object_key)
+        if file_contents is None:
+            raise Exception(f"File contents are None for {object_key}")
+            
+        # file_contents is a bytes object, decode it
+        file_contents = file_contents.decode("utf-8")
+        
+        # Ensure directory exists
+        os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
+        
+        # Write to local file
+        with open(local_file_path, 'w') as f:
+            f.write(file_contents)
+            
+        verbose_proxy_logger.debug(f"Python file downloaded successfully to {local_file_path}")
+        return True
+
+    except Exception as e:
+        verbose_proxy_logger.error(f"Error downloading Python file from GCS: {str(e)}")
+        return False
+
+
 # # Example usage
 # bucket_name = 'litellm-proxy'
 # object_key = 'litellm_proxy_config.yaml'