From a60c724ae130915533765a71d2c16cf3f5505ee4 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Thu, 13 Mar 2025 16:31:24 +0000 Subject: [PATCH] Make everythong asynch, prepare for streaming uplpaded file --- amqp/adapter/cleverthis_service_adapter.py | 42 +++--------------- amqp/model/uploaded_file.py | 21 +++++++++ amqp/service/AMQ_PyAdapter.log | 0 amqp/service/multipart_parser.py | 50 ++++++++++++++++++++++ amqp/service/stream_parser.py | 47 ++++++++++++++++++++ 5 files changed, 123 insertions(+), 37 deletions(-) create mode 100644 amqp/model/uploaded_file.py delete mode 100644 amqp/service/AMQ_PyAdapter.log create mode 100644 amqp/service/multipart_parser.py create mode 100644 amqp/service/stream_parser.py diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index 693aab1..13d0ad5 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -1,21 +1,18 @@ import asyncio import json import logging -import python_multipart -import io import os import re import threading from aiohttp import ClientSession, ClientResponse -from fastapi import HTTPException, UploadFile -from python_multipart import multipart -from python_multipart.multipart import Field, File +from fastapi import HTTPException from typing import Dict, List, Tuple, Callable, Coroutine, AnyStr, Any from urllib.parse import parse_qs from xml.etree import ElementTree from amqp.adapter.amq_response_factory import AMQResponseFactory +from amqp.service.multipart_parser import CleverMultiPartParser from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import AMQMessage, AMQResponse @@ -25,37 +22,6 @@ lock = threading.Lock() parallel_workers = os.getenv("PARALLEL_WORKERS", default=10) -class CleverMultiPartParser: - def __init__(self, message: AMQMessage): - self.form_data = {} - # 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) - except Exception as e: - logging.error(f"Error parsing multipart/form-data: {e}") - self.form_data = message.body() - 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) - - def parse_request_body(message: AMQMessage): """ Parses the HTTP POST request body based on the MIME type. @@ -294,6 +260,7 @@ class CleverThisServiceAdapter: :return: AMQResponse to be sent back to the API Gateway via AMQP """ return await self.handle_possible_form( + message, self.session.post( f"http://{self.service_host}:{self.service_port}{message.path}", data=message.body, @@ -302,7 +269,8 @@ class CleverThisServiceAdapter: auth_user ) - async def handle_possible_form(self, request_coroutine, auth_user: Any) -> AMQResponse: + async def handle_possible_form(self, message: AMQMessage, request_coroutine, auth_user: Any) -> AMQResponse: + args = parse_request_body(message) return await request_coroutine async def head(self, message: AMQMessage, auth_user: Any) -> AMQResponse: diff --git a/amqp/model/uploaded_file.py b/amqp/model/uploaded_file.py new file mode 100644 index 0000000..4d14346 --- /dev/null +++ b/amqp/model/uploaded_file.py @@ -0,0 +1,21 @@ +import io + + +class UploadedFile: + def __init__(self, filename: str = None): + self.filename = filename + self.body_b64 = io.BytesIO() + + +def from_bytes(input_bytes: bytes) -> UploadedFile: + """ + Builds the UploadedFile from its serialized (byte array) form. + :param input_bytes: serialized message + :return: UploadedFile instance + """ + prolog_len = (input_bytes[0] << 8) + input_bytes[1] + metadata = input_bytes[2:prolog_len + 2].decode() + body_b64 = input_bytes[prolog_len + 2:] + + return UploadedFile() + diff --git a/amqp/service/AMQ_PyAdapter.log b/amqp/service/AMQ_PyAdapter.log deleted file mode 100644 index e69de29..0000000 diff --git a/amqp/service/multipart_parser.py b/amqp/service/multipart_parser.py new file mode 100644 index 0000000..52406db --- /dev/null +++ b/amqp/service/multipart_parser.py @@ -0,0 +1,50 @@ +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) diff --git a/amqp/service/stream_parser.py b/amqp/service/stream_parser.py new file mode 100644 index 0000000..6997781 --- /dev/null +++ b/amqp/service/stream_parser.py @@ -0,0 +1,47 @@ +import json +import struct +import io +import pika + +class StreamParser: + def __init__(self): + self.buffer = io.BytesIO() # Buffer to accumulate chunks + self.metadata_length = None # Length of the metadata + self.metadata = None # Parsed JSON metadata + self.file_writer = None # File writer for the remaining data + + def process_chunk(self, chunk): + """ + Process a chunk of data and handle it according to the rules. + """ + self.buffer.write(chunk) + + # If metadata length is not yet known, try to read it + if self.metadata_length is None and self.buffer.tell() >= 4: + self.buffer.seek(0) + self.metadata_length = struct.unpack('>I', self.buffer.read(4))[0] + + # If metadata length is known but metadata is not yet parsed, try to parse it + if self.metadata_length is not None and self.metadata is None: + if self.buffer.tell() >= 4 + self.metadata_length: + self.buffer.seek(4) + metadata_bytes = self.buffer.read(self.metadata_length) + self.metadata = json.loads(metadata_bytes.decode('utf-8')) + print("Metadata:", self.metadata) + + # Prepare to write the remaining data to a file + self.file_writer = open(self.metadata.get("filename", "output_file"), "wb") + + # If metadata is parsed, write the remaining data to the file + if self.metadata is not None: + remaining_data = self.buffer.read() + if remaining_data: + self.file_writer.write(remaining_data) + + def close(self): + """ + Close the file writer and clean up. + """ + if self.file_writer: + self.file_writer.close() + self.buffer.close()