Stanislav Hejny
51 lines
2.0 KiB
Python
51 lines
2.0 KiB
Python
import io
|
|
import json
|
|
import logging
|
|
|
|
import python_multipart
|
|
from fastapi import UploadFile
|
|
from python_multipart import multipart
|
|
from python_multipart.multipart import Field, File
|
|
|
|
from amqp.model.model import AMQMessage
|
|
|
|
|
|
class CleverMultiPartParser:
|
|
def __init__(self, message: AMQMessage):
|
|
self.form_data = {}
|
|
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()}
|
|
content_type: str | bytes | None = headers.get("Content-Type")
|
|
content_type, params = multipart.parse_options_header(content_type)
|
|
if content_type.decode('utf-8') in ["application/octet-stream",
|
|
"multipart/form-data",
|
|
"application/x-www-form-urlencoded",
|
|
"application/x-url-encoded"]:
|
|
try:
|
|
python_multipart.parse_form(headers, io.BytesIO(message.body()), self.on_field, self.on_file)
|
|
self.is_multipart = True
|
|
except Exception as e:
|
|
logging.error(f"Error parsing multipart/form-data: {e}")
|
|
try:
|
|
self.form_data = json.loads(message.body())
|
|
self.is_json = True
|
|
except Exception as jsonerror:
|
|
logging.error(f"Error parsing JSON: {jsonerror}")
|
|
self.is_valid = False
|
|
else:
|
|
self.form_data = message.body()
|
|
|
|
def on_field(self, field: Field):
|
|
print(field)
|
|
self.form_data[field.field_name.decode('utf-8')] = field.value
|
|
print(self.form_data)
|
|
|
|
def on_file(self, file: File):
|
|
print(file)
|
|
self.form_data[file.field_name.decode('utf-8')] = \
|
|
UploadFile(file.file_object, size=file.size, filename=file.file_name)
|
|
print(self.form_data)
|