Merge pull request #29 from TaskarCenterAtUW/feature-sas-url

SAS URL generation done

Author: susrisha

src/python_ms_core/core/storage/abstract/storage_client.py

@@ -1,13 +1,71 @@
 from abc import ABC, abstractmethod
 
 
+
 class StorageClient(ABC):
+    """Abstract base class for a storage client."""
+
     @abstractmethod
-    def get_container(self, name: str): pass
+    def get_container(self, name: str):
+        """Get a container by name.
+
+        Args:
+            name (str): The name of the container.
+
+        Returns:
+            The container object.
+        """
+        pass
+
+    @abstractmethod
+    def get_file(self, container_name: str, file_name: str):
+        """Get a file from a container.
+
+        Args:
+            container_name (str): The name of the container.
+            file_name (str): The name of the file.
+
+        Returns:
+            The file object.
+        """
+        pass
 
     @abstractmethod
-    def get_file(self, container_name: str, file_name: str): pass
+    def get_file_from_url(self, container_name: str, full_url: str):
+        """Get a file from a URL.
+
+        Args:
+            container_name (str): The name of the container.
+            full_url (str): The full URL of the file.
+
+        Returns:
+            The file object.
+        """
+        pass
 
     @abstractmethod
-    def get_file_from_url(self, container_name: str, full_url: str): pass
+    def get_sas_url(self, container_name: str, file_path: str, expiry_hours: int) -> str:
+        """Get a shared access signature (SAS) URL for a file.
+
+        Args:
+            container_name (str): The name of the container.
+            file_path (str): The path of the file.
+            expiry_hours (int): The number of hours until the SAS URL expires.
+
+        Returns:
+            The SAS URL as a string.
+        """
+        pass
+# class StorageClient(ABC):
+#     #
+#     @abstractmethod
+#     def get_container(self, name: str): pass
+
+#     @abstractmethod
+#     def get_file(self, container_name: str, file_name: str): pass
+
+#     @abstractmethod
+#     def get_file_from_url(self, container_name: str, full_url: str): pass
 
+#     @abstractmethod
+#     def get_sas_url(self, container_name:str,file_path: str, expiry_hours:int) -> str: pass

src/python_ms_core/core/storage/providers/azure/azure_storage_client.py

@@ -1,12 +1,16 @@
 import logging
 import urllib.parse
-from azure.storage.blob import BlobServiceClient
+from azure.storage.blob import BlobServiceClient, generate_blob_sas, BlobSasPermissions
 from . import azure_file_entity, azure_storage_container
 from ...abstract import storage_client
 from ....resource_errors import ExceptionHandler
-
+import datetime
 
 class AzureStorageClient(storage_client.StorageClient):
+    """
+    Represents a client for interacting with Azure Storage.
+    """
+
     _blob_service_client: BlobServiceClient
 
     def __init__(self, config):
@@ -16,6 +20,15 @@ def __init__(self, config):
 
     @ExceptionHandler.decorated
     def get_container(self, container_name: str):
+        """
+        Retrieves the Azure Storage container with the specified name.
+
+        Args:
+            container_name (str): The name of the container.
+
+        Returns:
+            azure_storage_container.AzureStorageContainer: The Azure Storage container object.
+        """
         if container_name:
             client_container = self._blob_service_client.get_container_client(container_name)
             return azure_storage_container.AzureStorageContainer(container_name, client_container)
@@ -25,12 +38,32 @@ def get_container(self, container_name: str):
 
     @ExceptionHandler.decorated
     def get_file(self, container_name: str, file_name: str):
+        """
+        Retrieves the Azure Storage file with the specified name from the specified container.
+
+        Args:
+            container_name (str): The name of the container.
+            file_name (str): The name of the file.
+
+        Returns:
+            azure_file_entity.AzureFileEntity: The Azure Storage file object.
+        """
         client_container = self._blob_service_client.get_container_client(container_name)
         blob_client = client_container.get_blob_client(file_name)
         return azure_file_entity.AzureFileEntity(file_name, blob_client)
 
     @ExceptionHandler.decorated
     def get_file_from_url(self, container_name: str, full_url: str):
+        """
+        Retrieves the Azure Storage file from the specified URL.
+
+        Args:
+            container_name (str): The name of the container.
+            full_url (str): The full URL of the file.
+
+        Returns:
+            azure_file_entity.AzureFileEntity: The Azure Storage file object.
+        """
         path = '/'.join(urllib.parse.unquote(full_url).split('/')[4:])
         file = azure_file_entity.AzureFileEntity
         client_container = self._blob_service_client.get_container_client(container_name)
@@ -40,3 +73,27 @@ def get_file_from_url(self, container_name: str, full_url: str):
             if file_info.file_path == path:
                 file = file_info
         return file
+
+
+    def get_sas_url(self, container_name: str, file_path: str,  expiry_hours:int = 12) -> str:
+            """
+            Generates a Shared Access Signature (SAS) URL for the specified Azure Storage file.
+
+            Args:
+                container_name (str): The name of the container.
+                file_path (str): The path of the file within the container.
+                expiry_hours (int, optional): The number of hours until the SAS URL expires. Defaults to 12.
+
+            Returns:
+                str: The SAS URL for the file.
+            """
+            sas_token = generate_blob_sas(
+                self._blob_service_client.account_name,
+                container_name,
+                file_path,
+                account_key=self._blob_service_client.credential.account_key,
+                permission=BlobSasPermissions(read=True),
+                expiry=datetime.datetime.utcnow() + datetime.timedelta(hours=expiry_hours)
+            )
+            sas_url = f"https://{self._blob_service_client.account_name}.blob.core.windows.net/{container_name}/{file_path}?{sas_token}"
+            return sas_url

src/python_ms_core/core/storage/providers/local/local_storage_client.py

@@ -33,3 +33,7 @@ def get_file_from_url(self, container_name: str, full_url: str):
             if file_info.path == path:
                 file = file_info
         return file
+
+    @ExceptionHandler.decorated
+    def get_sas_url(self, container_name: str, file_path: str, expiry_hours: int) -> str:
+        return self.get_file_from_url(container_name, file_path).get_remote_url()