Skip to content

Instantly share code, notes, and snippets.

@tgwizard
Last active September 5, 2024 01:43
Show Gist options
  • Save tgwizard/95b82c98e17e72a4c3c0d75dda19eef4 to your computer and use it in GitHub Desktop.
Save tgwizard/95b82c98e17e72a4c3c0d75dda19eef4 to your computer and use it in GitHub Desktop.
A Flask request multipart/form-data file streamer
import time
import werkzeug.formparser
from flask import Flask, Response, request
class DummyWerkzeugFile:
def write(self, b: bytes):
print('reading file parts: size=%s' % len(b))
def seek(self, *args, **kwargs):
# Hack: this is how we know we've finished reading the request file.
return 0
def stream_factory(total_content_length, content_type, filename, content_length=None):
print(
'Start writing to stream: '
'total_content_length=%s, content_type=%s, filename=%s, content_length=%s' % (
total_content_length, content_type, filename, content_length
)
)
# Here we can return anything with a write(b: bytes) and seek() method, like a file.
return DummyWerkzeugFile()
def main():
app = Flask('file-streamer')
@app.route('/upload', methods=['POST'])
def upload():
print('Starting to read request')
start = time.time()
stream, form, files = werkzeug.formparser.parse_form_data(
request.environ, stream_factory=stream_factory
)
end = time.time()
print('Finished reading request: time=%s' % (end - start))
print('Form: %s' % form)
print('Files: %s' % files)
return Response(status=200)
app.run('0.0.0.0', 5001)
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment