-
Notifications
You must be signed in to change notification settings - Fork 2
Extract multipart parsing into _parse_multipart_files helper #2958
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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() | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Missing keyword-only argument enforcementMedium Severity The |
||
| 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) | ||
|
|
||


There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Duplicate multipart parsing logic across modules
Low Severity
The new
_parse_multipart_fileshelper duplicates multipart parsing logic already present inget_query_match_response_textin_query_tools.pyandvalidate_extra_fieldsinfields_validators.py. All three locations use identical code to extract boundary and parse multipart data. A shared utility function could eliminate this duplication and improve maintainability.