From 641a748616fcefb7a13009909e83a4816f6131ab Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Wed, 18 Feb 2026 15:30:25 +0000 Subject: [PATCH] Extract multipart parsing into _parse_multipart_files helper Refactor five image validators to eliminate duplicate multipart parsing code by extracting it into a reusable helper with @beartype decoration. Also simplify validate_image_is_image by directly accessing the stream from the files dict. Co-Authored-By: Claude Haiku 4.5 --- .../_query_validators/image_validators.py | 80 ++++++++++--------- 1 file changed, 41 insertions(+), 39 deletions(-) diff --git a/src/mock_vws/_query_validators/image_validators.py b/src/mock_vws/_query_validators/image_validators.py index 827c636d2..e2f8f498f 100644 --- a/src/mock_vws/_query_validators/image_validators.py +++ b/src/mock_vws/_query_validators/image_validators.py @@ -7,6 +7,7 @@ from beartype import beartype from PIL import Image +from werkzeug.datastructures import FileStorage, MultiDict from werkzeug.formparser import MultiPartParser from mock_vws._query_validators.exceptions import ( @@ -19,19 +20,19 @@ @beartype -def validate_image_field_given( +def _parse_multipart_files( *, request_headers: Mapping[str, str], request_body: bytes, -) -> None: - """Validate that the image field is given. +) -> MultiDict[str, FileStorage]: + """Parse the multipart body and return the files section. Args: request_headers: The headers sent with the request. request_body: The body of the request. - Raises: - ImageNotGivenError: The image field is not given. + Returns: + The files parsed from the multipart body. """ email_message = EmailMessage() email_message["Content-Type"] = request_headers["Content-Type"] @@ -42,6 +43,28 @@ def validate_image_field_given( boundary=boundary.encode(encoding="utf-8"), content_length=len(request_body), ) + return files + + +@beartype +def validate_image_field_given( + *, + request_headers: Mapping[str, str], + request_body: bytes, +) -> None: + """Validate that the image field is given. + + Args: + request_headers: The headers sent with the request. + request_body: The body of the request. + + Raises: + ImageNotGivenError: The image field is not given. + """ + files = _parse_multipart_files( + request_headers=request_headers, + request_body=request_body, + ) if files.get(key="image") is not None: return @@ -64,14 +87,9 @@ def validate_image_file_size( Raises: RequestEntityTooLargeError: The image file size is too large. """ - email_message = EmailMessage() - email_message["Content-Type"] = request_headers["Content-Type"] - boundary = email_message.get_boundary(failobj="") - parser = MultiPartParser() - _, files = parser.parse( - stream=io.BytesIO(initial_bytes=request_body), - boundary=boundary.encode(encoding="utf-8"), - content_length=len(request_body), + files = _parse_multipart_files( + request_headers=request_headers, + request_body=request_body, ) image_part = files["image"] image_value = image_part.stream.read() @@ -105,14 +123,9 @@ def validate_image_dimensions( BadImageError: The image is given and is not within the maximum width and height limits. """ - email_message = EmailMessage() - email_message["Content-Type"] = request_headers["Content-Type"] - boundary = email_message.get_boundary(failobj="") - parser = MultiPartParser() - _, files = parser.parse( - stream=io.BytesIO(initial_bytes=request_body), - boundary=boundary.encode(encoding="utf-8"), - content_length=len(request_body), + files = _parse_multipart_files( + request_headers=request_headers, + request_body=request_body, ) image_part = files["image"] image_value = image_part.stream.read() @@ -142,14 +155,9 @@ def validate_image_format( Raises: BadImageError: The image is given and is not either a PNG or a JPEG. """ - email_message = EmailMessage() - email_message["Content-Type"] = request_headers["Content-Type"] - boundary = email_message.get_boundary(failobj="") - parser = MultiPartParser() - _, files = parser.parse( - stream=io.BytesIO(initial_bytes=request_body), - boundary=boundary.encode(encoding="utf-8"), - content_length=len(request_body), + files = _parse_multipart_files( + request_headers=request_headers, + request_body=request_body, ) image_part = files["image"] pil_image = Image.open(fp=image_part.stream) @@ -175,17 +183,11 @@ def validate_image_is_image( Raises: BadImageError: Image data is given and it is not an image file. """ - email_message = EmailMessage() - email_message["Content-Type"] = request_headers["Content-Type"] - boundary = email_message.get_boundary(failobj="") - parser = MultiPartParser() - _, files = parser.parse( - stream=io.BytesIO(initial_bytes=request_body), - boundary=boundary.encode(encoding="utf-8"), - content_length=len(request_body), + files = _parse_multipart_files( + request_headers=request_headers, + request_body=request_body, ) - image_part = files["image"] - image_file = image_part.stream + image_file = files["image"].stream try: Image.open(fp=image_file)