feat: file stream from azure blob storage

commons/data_access_layer/file_stream.py

@@ -0,0 +1,27 @@
+import os
+from azure.storage.blob.blockblobservice import BlockBlobService
+
+ACCOUNT_KEY = os.environ.get('AZURE_STORAGE_ACCOUNT_KEY')
+
+class FileStream:
+    def __init__(self, account_name:str, container_name:str):
+        """
+        Initialize the FileStream object. which is used to get the file stream from Azure Blob Storage.
+        `account_name`: The name of the Azure Storage account.
+        `container_name`: The name of the Azure Storage container.
+        """
+        self.account_name = account_name
+        self.container_name = container_name
+        self.blob_service = BlockBlobService(account_name=self.account_name, account_key=ACCOUNT_KEY)
+
+    def get_file_stream(self, filename:str):
+        import tempfile 
+        try:
+            local_file = tempfile.NamedTemporaryFile()
+            self.blob_service.get_blob_to_stream(self.container_name, filename, stream=local_file)
+
+            local_file.seek(0)
+            return local_file
+        except Exception as e:
+            print(e)
+            return None

tests/commons/data_access_layer/file_stream_test.py

@@ -0,0 +1,15 @@
+import json
+
+from commons.data_access_layer.file_stream import FileStream
+
+fs = FileStream("storagefiles2","ioetfiles")
+
+def test_get_file_stream_return_file_when_enter_file_name():
+    result = fs.get_file_stream("activity.json")
+
+    assert len(json.load(result)) == 15
+
+def test_get_file_stream_return_None_when_not_enter_file_name_or_incorrect_name():
+    result = fs.get_file_stream("")
+
+    assert result == None