Use context manager for PIL Image.open to ensure proper resource cleanup

src/mock_vws/_query_validators/image_validators.py

@@ -130,11 +130,11 @@ def validate_image_dimensions(
     image_part = files["image"]
     image_value = image_part.stream.read()
     image_file = io.BytesIO(initial_bytes=image_value)
-    pil_image = Image.open(fp=image_file)
-    max_width = 30000
-    max_height = 30000
-    if pil_image.height <= max_height and pil_image.width <= max_width:
-        return
+    with Image.open(fp=image_file) as pil_image:
+        max_width = 30000
+        max_height = 30000
+        if pil_image.height <= max_height and pil_image.width <= max_width:
+            return
 
     _LOGGER.warning(msg="The image dimensions are too large.")
     raise BadImageError
@@ -160,10 +160,9 @@ def validate_image_format(
         request_body=request_body,
     )
     image_part = files["image"]
-    pil_image = Image.open(fp=image_part.stream)
-
-    if pil_image.format in {"PNG", "JPEG"}:
-        return
+    with Image.open(fp=image_part.stream) as pil_image:
+        if pil_image.format in {"PNG", "JPEG"}:
+            return
 
     _LOGGER.warning(msg="The image format is not PNG or JPEG.")
     raise BadImageError
@@ -191,7 +190,8 @@ def validate_image_is_image(
     image_file = files["image"].stream
 
     try:
-        Image.open(fp=image_file)
+        with Image.open(fp=image_file) as _:
+            pass
     except OSError as exc:
         _LOGGER.warning(msg="The image is not an image file.")
         raise BadImageError from exc

src/mock_vws/_services_validators/image_validators.py

@@ -40,13 +40,12 @@ def validate_image_integrity(*, request_body: bytes) -> None:
     decoded = decode_base64(encoded_data=image)
 
     image_file = io.BytesIO(initial_bytes=decoded)
-    pil_image = Image.open(fp=image_file)
-
-    try:
-        pil_image.verify()
-    except SyntaxError as exc:
-        _LOGGER.warning(msg="The image is not a valid image file.")
-        raise BadImageError from exc
+    with Image.open(fp=image_file) as pil_image:
+        try:
+            pil_image.verify()
+        except SyntaxError as exc:
+            _LOGGER.warning(msg="The image is not a valid image file.")
+            raise BadImageError from exc
 
 
 @beartype
@@ -70,10 +69,9 @@ def validate_image_format(*, request_body: bytes) -> None:
 
     decoded = decode_base64(encoded_data=image)
     image_file = io.BytesIO(initial_bytes=decoded)
-    pil_image = Image.open(fp=image_file)
-
-    if pil_image.format in {"PNG", "JPEG"}:
-        return
+    with Image.open(fp=image_file) as pil_image:
+        if pil_image.format in {"PNG", "JPEG"}:
+            return
 
     _LOGGER.warning(msg="The image is not a PNG or JPEG.")
     raise BadImageError
@@ -101,10 +99,9 @@ def validate_image_color_space(*, request_body: bytes) -> None:
 
     decoded = decode_base64(encoded_data=image)
     image_file = io.BytesIO(initial_bytes=decoded)
-    pil_image = Image.open(fp=image_file)
-
-    if pil_image.mode in {"L", "RGB"}:
-        return
+    with Image.open(fp=image_file) as pil_image:
+        if pil_image.mode in {"L", "RGB"}:
+            return
 
     _LOGGER.warning(
         msg="The image is not in the RGB or greyscale color space.",
@@ -165,7 +162,8 @@ def validate_image_is_image(*, request_body: bytes) -> None:
     image_file = io.BytesIO(initial_bytes=decoded)
 
     try:
-        Image.open(fp=image_file)
+        with Image.open(fp=image_file) as _:
+            pass
     except OSError as exc:
         raise BadImageError from exc
 

src/mock_vws/image_matchers.py

@@ -69,14 +69,16 @@ def __call__(
             second_image_content: Another image's content.
         """
         first_image_file = io.BytesIO(initial_bytes=first_image_content)
-        first_image = Image.open(fp=first_image_file)
         second_image_file = io.BytesIO(initial_bytes=second_image_content)
-        second_image = Image.open(fp=second_image_file)
-        # Images must be the same size, and they must be larger than the
-        # default SSIM window size of 11x11.
-        target_size = (256, 256)
-        first_image_resized = first_image.resize(size=target_size)
-        second_image_resized = second_image.resize(size=target_size)
+        with (
+            Image.open(fp=first_image_file) as first_image,
+            Image.open(fp=second_image_file) as second_image,
+        ):
+            # Images must be the same size, and they must be larger than the
+            # default SSIM window size of 11x11.
+            target_size = (256, 256)
+            first_image_resized = first_image.resize(size=target_size)
+            second_image_resized = second_image.resize(size=target_size)
 
         first_image_np = np.array(object=first_image_resized, dtype=np.float32)
         first_image_tensor = torch.tensor(data=first_image_np).float() / 255

src/mock_vws/target.py

@@ -92,10 +92,9 @@ def _post_processing_status(self) -> TargetStatuses:
         suitable the target is for detection.
         """
         image_file = io.BytesIO(initial_bytes=self.image_value)
-        image = Image.open(fp=image_file)
-        image_stat = ImageStat.Stat(image_or_list=image)
-
-        average_std_dev = statistics.mean(data=image_stat.stddev)
+        with Image.open(fp=image_file) as image:
+            image_stat = ImageStat.Stat(image_or_list=image)
+            average_std_dev = statistics.mean(data=image_stat.stddev)
 
         success_threshold = 5
 

src/mock_vws/target_raters.py

@@ -26,14 +26,14 @@ def _get_brisque_target_tracking_rating(*, image_content: bytes) -> int:
         image_content: A target's image's content.
     """
     image_file = io.BytesIO(initial_bytes=image_content)
-    image = Image.open(fp=image_file)
-    image_np = np.array(object=image, dtype=np.float32)
-    image_tensor = torch.tensor(data=image_np).float() / 255
-    image_tensor = image_tensor.view(
-        image.size[1],
-        image.size[0],
-        len(image.getbands()),
-    )
+    with Image.open(fp=image_file) as image:
+        image_np = np.array(object=image, dtype=np.float32)
+        image_tensor = torch.tensor(data=image_np).float() / 255
+        image_tensor = image_tensor.view(
+            image.size[1],
+            image.size[0],
+            len(image.getbands()),
+        )
     image_tensor = image_tensor.permute(2, 0, 1).unsqueeze(dim=0)
     try:
         brisque_score = brisque(x=image_tensor, data_range=255)