streaming-form-data

import io from streaming_form_data import StreamingFormDataParser from streaming_form_data.targets import ValueTarget, FileTarget # Simulate an incoming HTTP request body and headers boundary = b'----WebKitFormBoundary7MA4YWxkTrZu0gW' content_type = b'multipart/form-data; boundary=' + boundary payload = ( b'--' + boundary + b'\r\n' b'Content-Disposition: form-data; name="text_field"\r\n' b'\r\n' b'Hello Streaming World!' b'\r\n' b'--' + boundary + b'\r\n' b'Content-Disposition: form-data; name="file_upload"; filename="message.txt"\r\n' b'Content-Type: text/plain\r\n' b'\r\n' b'This is a test file content.\nLine two.\n' b'\r\n' b'--' + boundary + b'--\r\n' ) headers = {b'Content-Type': content_type} parser = StreamingFormDataParser(headers=headers) text_field = ValueTarget() file_target = FileTarget(file_path='/tmp/uploaded_message.txt') # Path where the file will be saved parser.register('text_field', text_field) parser.register('file_upload', file_target) # In a real web application, `request.body` would be streamed here # For this example, we use a BytesIO object to simulate the stream stream = io.BytesIO(payload) while True: chunk = stream.read(8192) # Read in chunks if not chunk: break parser.data_received(chunk) print(f"Text field value: '{text_field.value.decode()}'") print(f"File saved to: '{file_target.file_path}'") # Verify content (optional, for demonstration) with open(file_target.file_path, 'rb') as f: print(f"File content: '{f.read().decode()}'")