yetangitu/photo-python
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Merge pull request #19 from sneakypete81/multipart

Browse source 
Multipart Form Upload Support
This commit is contained in:
sneakypete81 2013-04-02 13:05:08 -07:00
commit 816887ff06
6 changed files with 95 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

4
openphoto/api_photo.py
View file

@ -67,7 +67,9 @@ class ApiPhoto:
return photo
def upload(self, photo_file, **kwds):
raise NotImplementedError("Use upload_encoded instead.")
result = self._client.post("/photo/upload.json", files={'photo': photo_file},
**kwds)["result"]
return Photo(self._client, result)
def upload_encoded(self, photo_file, **kwds):
""" Base64-encodes and uploads the specified file """

22
openphoto/main.py
View file

@ -47,7 +47,8 @@ def main(args=sys.argv[1:]):
if options.method == "GET":
result = client.get(options.endpoint, process_response=False, **params)
else:
result = client.post(options.endpoint, process_response=False, **params)
params, files = extract_files(params)
result = client.post(options.endpoint, process_response=False, files=files, **params)
if options.verbose:
print "==========\nMethod: %s\nHost: %s\nEndpoint: %s" % (options.method, options.host, options.endpoint)
@ -62,5 +63,24 @@ def main(args=sys.argv[1:]):
else:
print result
def extract_files(params):
"""
Extract filenames from the "photo" parameter, so they can be uploaded, returning (updated_params, files).
Uses the same technique as openphoto-php:
* Filename can only be in the "photo" parameter
* Filename must be prefixed with "@"
* Filename must exist
...otherwise the parameter is not extracted
"""
files = {}
updated_params = {}
for name in params:
if name == "photo" and params[name].startswith("@") and os.path.isfile(os.path.expanduser(params[name][1:])):
files[name] = params[name][1:]
else:
updated_params[name] = params[name]
return updated_params, files
if __name__ == "__main__":
main()

31
openphoto/multipart_post.py Normal file
View file

@ -0,0 +1,31 @@
import os
import mimetypes
import mimetools
def encode_multipart_formdata(params, files):
boundary = mimetools.choose_boundary()
lines = []
for name in params:
lines.append("--" + boundary)
lines.append("Content-Disposition: form-data; name=\"%s\"" % name)
lines.append("")
lines.append(str(params[name]))
for name in files:
filename = files[name]
content_type, _ = mimetypes.guess_type(filename)
if content_type is None:
content_type = "application/octet-stream"
lines.append("--" + boundary)
lines.append("Content-Disposition: form-data; name=\"%s\"; filename=\"%s\"" % (name, filename))
lines.append("Content-Type: %s" % content_type)
lines.append("")
lines.append(open(os.path.expanduser(filename), "rb").read())
lines.append("--" + boundary + "--")
lines.append("")
body = "\r\n".join(lines)
headers = {'Content-Type': "multipart/form-data; boundary=%s" % boundary,
'Content-Length': str(len(body))}
return headers, body

28
openphoto/openphoto_http.py
View file

@ -1,6 +1,7 @@
import oauth2 as oauth
import urlparse
import urllib
import urllib2
import httplib2
try:
import json
@ -9,6 +10,7 @@ except ImportError:
from objects import OpenPhotoObject
from errors import *
from multipart_post import encode_multipart_formdata
DUPLICATE_RESPONSE = {"code": 409,
"message": "This photo already exists"}
@ -58,7 +60,7 @@ class OpenPhotoHttp:
else:
return content
def post(self, endpoint, process_response=True, **params):
def post(self, endpoint, process_response=True, files = {}, **params):
"""
Performs an HTTP POST to the specified endpoint (API path),
passing parameters if given.
@ -74,10 +76,17 @@ class OpenPhotoHttp:
consumer = oauth.Consumer(self._consumer_key, self._consumer_secret)
token = oauth.Token(self._token, self._token_secret)
client = oauth.Client(consumer, token)
body = urllib.urlencode(params)
_, content = client.request(url, "POST", body)
if files:
# Parameters must be signed and encoded into the multipart body
params = self._sign_params(client, url, params)
headers, body = encode_multipart_formdata(params, files)
request = urllib2.Request(url, body, headers)
content = urllib2.urlopen(request).read()
else:
body = urllib.urlencode(params)
_, content = client.request(url, "POST", body)
self.last_url = url
self.last_params = params
@ -88,6 +97,17 @@ class OpenPhotoHttp:
else:
return content
@staticmethod
def _sign_params(client, url, params):
"""Use OAuth to sign a dictionary of params"""
request = oauth.Request.from_consumer_and_token(consumer=client.consumer,
token=client.token,
http_method="POST",
http_url=url,
parameters=params)
request.sign_request(client.method, client.consumer, client.token)
return dict(urlparse.parse_qsl(request.to_postdata()))
@staticmethod
def _process_params(params):
""" Converts Unicode/lists/booleans inside HTTP parameters """

18
tests/test_base.py
View file

@ -100,15 +100,15 @@ class TestBase(unittest.TestCase):
""" Upload three test photos """
album = cls.client.album.create(cls.TEST_ALBUM)
photos = [
cls.client.photo.upload_encoded("tests/test_photo1.jpg",
title=cls.TEST_TITLE,
albums=album.id),
cls.client.photo.upload_encoded("tests/test_photo2.jpg",
title=cls.TEST_TITLE,
albums=album.id),
cls.client.photo.upload_encoded("tests/test_photo3.jpg",
title=cls.TEST_TITLE,
albums=album.id),
cls.client.photo.upload("tests/test_photo1.jpg",
title=cls.TEST_TITLE,
albums=album.id),
cls.client.photo.upload("tests/test_photo2.jpg",
title=cls.TEST_TITLE,
albums=album.id),
cls.client.photo.upload("tests/test_photo3.jpg",
title=cls.TEST_TITLE,
albums=album.id),
]
# Add the test tag, removing any autogenerated tags
for photo in photos:

19
tests/test_photos.py
View file

@ -15,11 +15,11 @@ class TestPhotos(test_base.TestBase):
# Check that they're gone
self.assertEqual(self.client.photos.list(), [])
# Re-upload the photos
ret_val = self.client.photo.upload_encoded("tests/test_photo1.jpg",
title=self.TEST_TITLE)
self.client.photo.upload_encoded("tests/test_photo2.jpg",
title=self.TEST_TITLE)
# Re-upload the photos, one of them using Bas64 encoding
ret_val = self.client.photo.upload("tests/test_photo1.jpg",
title=self.TEST_TITLE)
self.client.photo.upload("tests/test_photo2.jpg",
title=self.TEST_TITLE)
self.client.photo.upload_encoded("tests/test_photo3.jpg",
title=self.TEST_TITLE)
@ -56,8 +56,8 @@ class TestPhotos(test_base.TestBase):
""" Ensure that duplicate photos are rejected """
# Attempt to upload a duplicate
with self.assertRaises(openphoto.OpenPhotoDuplicateError):
self.client.photo.upload_encoded("tests/test_photo1.jpg",
title=self.TEST_TITLE)
self.client.photo.upload("tests/test_photo1.jpg",
title=self.TEST_TITLE)
# Check there are still three photos
self.photos = self.client.photos.list()
@ -141,11 +141,6 @@ class TestPhotos(test_base.TestBase):
with self.assertRaises(openphoto.NotImplementedError):
self.client.photo.replace_encoded(None, None)
def test_upload(self):
""" If photo.upload gets implemented, write a test! """
with self.assertRaises(openphoto.NotImplementedError):
self.client.photo.upload(None)
def test_dynamic_url(self):
""" If photo.dynamic_url gets implemented, write a test! """
with self.assertRaises(openphoto.NotImplementedError):