Stanislav Hejny
195 lines
8.1 KiB
Python
195 lines
8.1 KiB
Python
import io
|
|
import json
|
|
from typing import Dict, Tuple
|
|
from urllib.parse import parse_qs, urlparse
|
|
from xml.etree import ElementTree
|
|
|
|
import python_multipart
|
|
from fastapi import UploadFile
|
|
from python_multipart.multipart import Field, File
|
|
|
|
from amqp.adapter.logging_utils import logging_debug, logging_error
|
|
from amqp.model.model import AMQMessage
|
|
|
|
|
|
class MultipartFormDataParser:
|
|
"""
|
|
Parses multipart/form-data content and extracts form fields (key/value pairs) and files.
|
|
Note: multipart message that contains file(s) to upload are converted to CleverMicro JSON-based streaming format,
|
|
where files are transmitted separately via stream using dedicated queue, so the message content is no longer
|
|
the HTTP multipart format, but the Content-Type is being preserved for compatibility with the original message.
|
|
This class supports both formats (multipart and JSON) depending on the message content, because if no file is
|
|
included, the message is transmitted in the unchanged multipart/form-data format.
|
|
"""
|
|
|
|
def __init__(self, message: AMQMessage):
|
|
self.form_data: dict = {}
|
|
self.is_multipart = False
|
|
self.is_json = False
|
|
self.is_valid = True
|
|
# Convert each list of strings to a single string joined by newline, then to bytes
|
|
_headers = {
|
|
key: "\n".join(value).encode("utf-8") for key, value in message.headers().items()
|
|
}
|
|
try:
|
|
if (
|
|
hasattr(message, "extra_data")
|
|
and isinstance(message.extra_data, dict)
|
|
and len(message.extra_data) > 0
|
|
):
|
|
try:
|
|
self.form_data = json.loads(message.body())
|
|
self.is_json = True
|
|
if self.form_data["files"] is not None and self.form_data["form"] is not None:
|
|
self.form_data = AMQDataParser.process_form_data(self.form_data)
|
|
except Exception as jsonerror:
|
|
logging_error(f"Parsing JSON: {jsonerror}")
|
|
self.is_valid = False
|
|
else:
|
|
python_multipart.parse_form(
|
|
_headers, io.BytesIO(message.body()), self.on_field, self.on_file
|
|
)
|
|
self.is_multipart = True
|
|
except Exception as e:
|
|
__error = f"Error parsing multipart/form-data: {e}. Trying parsing as JSON."
|
|
try:
|
|
self.form_data = json.loads(message.body())
|
|
self.is_json = True
|
|
if self.form_data["files"] is not None and self.form_data["form"] is not None:
|
|
self.form_data = AMQDataParser.process_form_data(self.form_data)
|
|
except Exception as jsonerror:
|
|
logging_error(__error)
|
|
logging_error(f"Parsing JSON: {jsonerror}")
|
|
self.is_valid = False
|
|
|
|
def on_field(self, field: Field):
|
|
logging_debug(str(field))
|
|
self.form_data[field.field_name.decode("utf-8")] = field.value
|
|
|
|
def on_file(self, file: File):
|
|
logging_debug(str(file))
|
|
file.file_object.seek(0)
|
|
self.form_data[file.field_name.decode("utf-8")] = UploadFile(
|
|
file.file_object, size=file.size, filename=file.file_name.decode("utf-8")
|
|
)
|
|
|
|
|
|
class AMQDataParser:
|
|
"""
|
|
A class for parsing AMQ data messages. It provides methods to parse form data,
|
|
"""
|
|
|
|
@staticmethod
|
|
def process_form_data(json_input) -> dict:
|
|
"""
|
|
Processes the input JSON to extract form data. The JSON has two top entries: 'files' and 'form'.
|
|
This is part of the file stream implementation in AMQ adapter to stream large files in chunks.
|
|
'files' contains information about the AMQ queue from which read the file, while 'form' contains
|
|
additional non-file form values.
|
|
This function combines the file and values at the top level in the returned dictionary.
|
|
Args:
|
|
json_input: JSON string or dict containing the form and files data
|
|
|
|
Returns:
|
|
dict: Processed form data with files data overriding non-empty values
|
|
"""
|
|
# Load JSON if input is string
|
|
if isinstance(json_input, str):
|
|
data = json.loads(json_input)
|
|
else:
|
|
data = json_input
|
|
|
|
result = {}
|
|
|
|
# Process form data - take first element of each list
|
|
for key, value in data["form"].items():
|
|
if value: # Only process if list is not empty
|
|
result[key] = value[0]
|
|
|
|
# Override with files data where available
|
|
for key, value in data["files"].items():
|
|
if value: # Only override if files data is non-empty
|
|
result[key] = value[0]
|
|
return result
|
|
|
|
@staticmethod
|
|
def parse_get_url(url: str) -> Tuple[str, Dict[str, str]]:
|
|
"""
|
|
Parses an HTTP GET URL and returns (context_path, query_params_dict)
|
|
|
|
Args:
|
|
url: The URL string to parse (e.g., "http://example.com/path?foo=1&bar=2&foo=3")
|
|
|
|
Returns:
|
|
tuple: (context_path, query_params_dict)
|
|
- context_path: The path part of the URL (e.g., "/path")
|
|
- query_params_dict: Query parameters where values are str for single value or list of str for multivalue
|
|
"""
|
|
_parsed = urlparse(url)
|
|
_query_params = parse_qs(_parsed.query)
|
|
# Convert values from lists to single values when there's only one value
|
|
_simplified_params = {k: v[0] if len(v) == 1 else v for k, v in _query_params.items()}
|
|
return _parsed.path, _simplified_params
|
|
|
|
@staticmethod
|
|
def parse_request_body(message: AMQMessage):
|
|
"""
|
|
Parses the HTTP POST request body based on the MIME type.
|
|
This is typically called fom the concrete CleverThis Service to extract parameters from the message body.
|
|
Args:
|
|
message (DataMessage): Contains raw request and MIME headers.
|
|
|
|
Returns:
|
|
dict: Parsed data as a dictionary.
|
|
"""
|
|
_content_type = message.headers().get("Content-Type", "")
|
|
if not _content_type:
|
|
raise ValueError("Content-Type header is required")
|
|
content_type = _content_type[0]
|
|
# Parse JSON
|
|
if content_type == "application/json":
|
|
try:
|
|
return json.loads(message.body() or "{}")
|
|
except json.JSONDecodeError as e:
|
|
logging_error(f"Invalid JSON: {e}")
|
|
raise ValueError(f"Invalid JSON: {e}")
|
|
|
|
# Parse URL-encoded form data
|
|
elif content_type == "application/x-www-form-urlencoded":
|
|
try:
|
|
_form_data: dict = {
|
|
k: v[0] if len(v) == 1 else v for k, v in parse_qs(message.body()).items()
|
|
}
|
|
if len(_form_data) == 0 and len(message.body()) > 0:
|
|
_form_data = json.loads(message.body())
|
|
if _form_data["files"] is not None and _form_data["form"] is not None:
|
|
_form_data = AMQDataParser.process_form_data(_form_data)
|
|
return _form_data
|
|
except Exception as e:
|
|
logging_error(f"Invalid URL-encoded form data: {e}")
|
|
raise ValueError(f"Invalid URL-encoded form data: {e}")
|
|
|
|
# Parse multipart/form-data
|
|
elif content_type.startswith("multipart/form-data"):
|
|
try:
|
|
parser: MultipartFormDataParser = MultipartFormDataParser(message)
|
|
return parser.form_data
|
|
except Exception as e:
|
|
logging_error(f"Invalid multipart/form-data: {e}")
|
|
raise ValueError(f"Invalid multipart/form-data: {e}")
|
|
|
|
# Parse plain text
|
|
elif content_type == "text/plain":
|
|
return {"text": message.body()}
|
|
|
|
# Parse XML
|
|
elif content_type == "application/xml":
|
|
try:
|
|
_root = ElementTree.fromstring(message.body())
|
|
return {elem.tag: elem.text for elem in _root}
|
|
except ElementTree.ParseError as e:
|
|
raise ValueError(f"Invalid XML: {e}")
|
|
|
|
else:
|
|
raise ValueError(f"Unsupported Content-Type: {content_type}")
|