Update GCS image compression headers.

app/images/__init__.py

@@ -64,22 +64,26 @@ async def bulk_transform_images(
 
     :returns: JSONResponse
     """
-    if directory is None:
-        directory = settings.GCP_BUCKET_FOLDER
-    transformed_images = {
-        "purged": images.purge_unwanted_images(directory),
-        "retina": images.retina_transformations(directory),
-        "mobile": images.mobile_transformations(directory),
-        # "standard": gcs.standard_transformations(directory),
-    }
-    response = []
-    for k, v in transformed_images.items():
-        if v is not None:
-            response.append(f"{len(v)} {k}")
-        else:
-            response.append(f"0 {k}")
-    LOGGER.success(f"Transformed {', '.join(response)} images")
-    return JSONResponse(transformed_images)
+    try:
+        if directory is None:
+            directory = settings.GCP_BUCKET_FOLDER
+        transformed_images = {
+            "purged": images.purge_unwanted_images(directory),
+            "retina": images.retina_transformations(directory),
+            "mobile": images.mobile_transformations(directory),
+            # "standard": gcs.standard_transformations(directory),
+        }
+        response = []
+        for k, v in transformed_images.items():
+            if v is not None:
+                response.append(f"{len(v)} {k}")
+            else:
+                response.append(f"0 {k}")
+        LOGGER.success(f"Transformed {', '.join(response)} images")
+        return JSONResponse(transformed_images)
+    except Exception as e:
+        LOGGER.error(f"Unexpected exception raised when transforming images: {e}")
+        return JSONResponse(e, status_code=500)
 
 
 @router.get("/sort/")
@@ -94,11 +98,8 @@ async def bulk_organize_images(directory: Optional[str] = None) -> JSONResponse:
     if directory is None:
         directory = settings.GCP_BUCKET_FOLDER
     retina_images = images.organize_retina_images(directory)
-    image_headers = images.image_headers(directory)
-    LOGGER.success(f"Moved {len(retina_images)} retina images, modified {len(image_headers)} content types.")
+    LOGGER.success(f"Moved {len(retina_images)} retina images.")
     return JSONResponse(
-        {
-            "retina": retina_images,
-            "headers": image_headers,
-        }
+        {"retina": retina_images},
+        status_code=200,
     )

app/posts/__init__.py

@@ -37,8 +37,8 @@ async def update_post(post_update: PostUpdate) -> JSONResponse:
         current_time = get_current_datetime()
         previous_update_date = datetime.strptime(str(previous_update.updated_at), "%Y-%m-%dT%H:%M:%S.000Z")
         if previous_update_date and current_time - previous_update_date < timedelta(seconds=5):
-            LOGGER.warning("Post update ignored as post was just updated.")
-            raise HTTPException(status_code=422, detail="Post update ignored as post was just updated.")
+            LOGGER.warning("Post update ignored (post was recently updated).")
+            raise HTTPException(status_code=422, detail="Post update ignored (post was recently updated).")
     post = post_update.post.current
     slug = post.slug
     feature_image = post.feature_image

clients/img.py

@@ -148,9 +148,7 @@ def create_retina_image(self, image_blob: Blob) -> Optional[Blob]:
             new_retina_image_blob = self.bucket.blob(retina_blob_filepath)
             LOGGER.success(f"Created retina image `{retina_blob_filepath}`")
             return new_retina_image_blob
-        else:
-            LOGGER.info(f"Skipping retina image `{retina_blob_filepath}`; already exists.")
-        return None
+        LOGGER.info(f"Skipping retina image `{retina_blob_filepath}`; already exists.")
 
     @LOGGER.catch
     def mobile_transformations(self, folder: str) -> List[Optional[str]]:
@@ -186,23 +184,23 @@ def create_mobile_image(self, image_blob: Blob) -> Optional[Blob]:
         if mobile_image_blob.exists() is False:
             new_mobile_image_blob = self._transform_mobile_image(image_blob, mobile_image_blob)
             return new_mobile_image_blob
-        else:
-            LOGGER.info(f"Skipping mobile image `{mobile_blob_filepath}`; already exists.")
-        return None
+        LOGGER.info(f"Skipping mobile image `{mobile_blob_filepath}`; already exists.")
 
     @staticmethod
-    def _get_image_meta(blob: Blob) -> Optional[dict]:
+    def _set_image_metadata(blob: Blob) -> Optional[dict]:
         """
         Generate metadata for a given image Blob.
 
         :param Blob blob: Image blob being transformed.
 
         :returns: Optional[dict]
         """
-        if ".jpg" in blob.name:
-            return {"format": "JPEG", "content-type": "image/jpg"}
-        elif ".png" in blob.name:
+        if ".jpg" in blob.name and "octet-stream" in blob.content_type:
+            return {"format": "JPG", "content-type": "image/jpg"}
+        if ".png" in blob.name:
             return {"format": "PNG", "content-type": "image/png"}
+        if ".webp" in blob.name:
+            return {"format": "WEBP", "content-type": "image/webp"}
         return None
 
     def _transform_mobile_image(self, original_image_blob: Blob, new_image_blob: Blob) -> Optional[Blob]:
@@ -214,7 +212,7 @@ def _transform_mobile_image(self, original_image_blob: Blob, new_image_blob: Blo
 
         :returns: Optional[Blob]
         """
-        img_meta = self._get_image_meta(original_image_blob)
+        img_meta = self._set_image_metadata(original_image_blob)
         img_bytes = original_image_blob.download_as_bytes()
         if img_bytes:
             stream = BytesIO(img_bytes)
@@ -230,11 +228,3 @@ def _transform_mobile_image(self, original_image_blob: Blob, new_image_blob: Blo
                 LOGGER.error(f"GoogleCloudError while saving mobile image `{new_image_blob.name}`: {e}")
             except Exception as e:
                 LOGGER.error(f"Unexpected exception while saving mobile image `{new_image_blob.name}`: {e}")
-
-    @staticmethod
-    def _add_image_headers(image_blob: Blob):
-        if ".jpg" in image_blob.name and "octet-stream" in image_blob.content_type:
-            image_blob.content_type = "image/jpg"
-        elif ".png" in image_blob.name and "octet-stream" in image_blob.content_type:
-            image_blob.content_type = "image/png"
-        return image_blob