Allows manifest.json to be gzip compressed

dbt_loom/clients/az_blob.py

@@ -1,5 +1,7 @@
 import json
 import os
+import gzip
+from io import BytesIO
 from typing import Dict
 
 from azure.identity import DefaultAzureCredential
@@ -49,7 +51,11 @@ def load_manifest(self) -> Dict:
 
         # Deserialize the body of the object.
         try:
-            content = blob_client.download_blob(encoding="utf-8").readall()
+            if self.object_name.endswith('.gz'):
+                with gzip.GzipFile(fileobj=BytesIO(blob_client.download_blob().readall())) as gzipfile:
+                    content = gzipfile.read().decode('utf-8')
+            else:
+                content = blob_client.download_blob(encoding="utf-8").readall()
         except Exception:
             raise Exception(
                 f"Unable to read the data contained in the object `{self.object_name}"

dbt_loom/clients/gcs.py

@@ -1,4 +1,6 @@
 import json
+import gzip
+from io import BytesIO
 from pathlib import Path
 from typing import Dict, Optional
 
@@ -47,7 +49,12 @@ def load_manifest(self) -> Dict:
                 f"`{self.bucket_name}`."
             )
 
-        manifest_json = blob.download_as_text()
+        if self.object_name.endswith('.gz'):
+            compressed_manifest = blob.download_as_bytes()
+            with gzip.GzipFile(fileobj=BytesIO(compressed_manifest)) as gzip_file:
+                manifest_json = gzip_file.read()
+        else:
+            manifest_json = blob.download_as_text()
 
         try:
             return json.loads(manifest_json)

dbt_loom/clients/s3.py

@@ -3,6 +3,8 @@
 from typing import Dict, Optional
 
 import boto3
+import gzip
+from io import BytesIO
 from pydantic import BaseModel
 
 
@@ -39,7 +41,12 @@ def load_manifest(self) -> Dict:
 
         # Deserialize the body of the object.
         try:
-            content = response["Body"].read().decode("utf-8")
+            if self.object_name.endswith(".gz"):
+                body = response["Body"].read()
+                with gzip.GzipFile(fileobj=BytesIO(body)) as gzipfile:
+                    content = gzipfile.read().decode('utf-8')
+            else:
+                content = response["Body"].read().decode("utf-8")
         except Exception:
             raise Exception(
                 f"Unable to read the data contained in the object `{self.object_name}"

dbt_loom/manifests.py

@@ -1,5 +1,6 @@
 import datetime
 import json
+import gzip
 from typing import Dict, List, Optional
 
 from pydantic import BaseModel, Field, validator
@@ -80,8 +81,12 @@ def load_from_local_filesystem(config: FileReferenceConfig) -> Dict:
         """Load a manifest dictionary from a local file"""
         if not config.path.exists():
             raise LoomConfigurationError(f"The path `{config.path}` does not exist.")
-
-        return json.load(open(config.path))
+
+        if config.path.suffix == '.gz':
+            with gzip.open(config.path, 'rt') as file:
+                return json.load(file)
+        else:
+            return json.load(open(config.path))
 
     @staticmethod
     def load_from_dbt_cloud(config: DbtCloudReferenceConfig) -> Dict: