115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
|
# https://pymotw.com/3/urllib.request/#uploading-files
|
||
|
|
|
||
|
|
import io
|
||
|
|
import mimetypes
|
||
|
|
from urllib import request
|
||
|
|
import uuid
|
||
|
|
|
||
|
|
|
||
|
|
class MultiPartForm:
|
||
|
|
"""Accumulate the data to be used when posting a form."""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
self.form_fields = []
|
||
|
|
self.files = []
|
||
|
|
# Use a large random byte string to separate
|
||
|
|
# parts of the MIME data.
|
||
|
|
self.boundary = uuid.uuid4().hex.encode('utf-8')
|
||
|
|
return
|
||
|
|
|
||
|
|
def get_content_type(self):
|
||
|
|
return 'multipart/form-data; boundary={}'.format(
|
||
|
|
self.boundary.decode('utf-8'))
|
||
|
|
|
||
|
|
def add_field(self, name, value):
|
||
|
|
"""Add a simple field to the form data."""
|
||
|
|
self.form_fields.append((name, value))
|
||
|
|
|
||
|
|
def add_file(self, fieldname, filename, fileHandle,
|
||
|
|
mimetype=None):
|
||
|
|
"""Add a file to be uploaded."""
|
||
|
|
body = fileHandle.read()
|
||
|
|
if mimetype is None:
|
||
|
|
mimetype = (
|
||
|
|
mimetypes.guess_type(filename)[0] or
|
||
|
|
'application/octet-stream'
|
||
|
|
)
|
||
|
|
self.files.append((fieldname, filename, mimetype, body))
|
||
|
|
return
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _form_data(name):
|
||
|
|
return ('Content-Disposition: form-data; '
|
||
|
|
'name="{}"\r\n').format(name).encode('utf-8')
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _attached_file(name, filename):
|
||
|
|
return ('Content-Disposition: file; '
|
||
|
|
'name="{}"; filename="{}"\r\n').format(
|
||
|
|
name, filename).encode('utf-8')
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _content_type(ct):
|
||
|
|
return 'Content-Type: {}\r\n'.format(ct).encode('utf-8')
|
||
|
|
|
||
|
|
def __bytes__(self):
|
||
|
|
"""Return a byte-string representing the form data,
|
||
|
|
including attached files.
|
||
|
|
"""
|
||
|
|
buffer = io.BytesIO()
|
||
|
|
boundary = b'--' + self.boundary + b'\r\n'
|
||
|
|
|
||
|
|
# Add the form fields
|
||
|
|
for name, value in self.form_fields:
|
||
|
|
buffer.write(boundary)
|
||
|
|
buffer.write(self._form_data(name))
|
||
|
|
buffer.write(b'\r\n')
|
||
|
|
buffer.write(value.encode('utf-8'))
|
||
|
|
buffer.write(b'\r\n')
|
||
|
|
|
||
|
|
# Add the files to upload
|
||
|
|
for f_name, filename, f_content_type, body in self.files:
|
||
|
|
buffer.write(boundary)
|
||
|
|
buffer.write(self._attached_file(f_name, filename))
|
||
|
|
buffer.write(self._content_type(f_content_type))
|
||
|
|
buffer.write(b'\r\n')
|
||
|
|
buffer.write(body)
|
||
|
|
buffer.write(b'\r\n')
|
||
|
|
|
||
|
|
buffer.write(b'--' + self.boundary + b'--\r\n')
|
||
|
|
return buffer.getvalue()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
# Create the form with simple fields
|
||
|
|
form = MultiPartForm()
|
||
|
|
form.add_field('firstname', 'Doug')
|
||
|
|
form.add_field('lastname', 'Hellmann')
|
||
|
|
|
||
|
|
# Add a fake file
|
||
|
|
form.add_file(
|
||
|
|
'biography', 'bio.txt',
|
||
|
|
fileHandle=io.BytesIO(b'Python developer and blogger.'))
|
||
|
|
|
||
|
|
# Build the request, including the byte-string
|
||
|
|
# for the data to be posted.
|
||
|
|
data = bytes(form)
|
||
|
|
r = request.Request('http://localhost:8080/', data=data)
|
||
|
|
r.add_header(
|
||
|
|
'User-agent',
|
||
|
|
'PyMOTW (https://pymotw.com/)',
|
||
|
|
)
|
||
|
|
r.add_header('Content-type', form.get_content_type())
|
||
|
|
r.add_header('Content-length', len(data))
|
||
|
|
|
||
|
|
print()
|
||
|
|
print('OUTGOING DATA:')
|
||
|
|
for name, value in r.header_items():
|
||
|
|
print('{}: {}'.format(name, value))
|
||
|
|
print()
|
||
|
|
print(r.data.decode('utf-8'))
|
||
|
|
|
||
|
|
print()
|
||
|
|
print('SERVER RESPONSE:')
|
||
|
|
print(request.urlopen(r).read().decode('utf-8'))
|