Source code for amqp.adapter.data_parser
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
[docs]
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.
"""
[docs]
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
[docs]
def on_field(self, field: Field):
logging_debug(str(field))
self.form_data[field.field_name.decode("utf-8")] = field.value
[docs]
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")
)
[docs]
class AMQDataParser:
"""
A class for parsing AMQ data messages. It provides methods to parse form data,
"""
[docs]
@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
[docs]
@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
[docs]
@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}")