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

Merge branch 'python3' into development

Browse source
This commit is contained in:
sneakypete81 2013-06-29 11:23:23 +01:00
commit 1e22ba60d4
21 changed files with 463 additions and 424 deletions
Download patch file Download diff file Expand all files Collapse all files

2
README.markdown
View file

@ -53,7 +53,7 @@ You can use the library in one of two ways:
client = OpenPhoto()
photos = client.photos.list()
photos[0].update(tags=["tag1", "tag2"])
print photos[0].tags
print(photos[0].tags)
The OpenPhoto Python class hierarchy mirrors the [OpenPhoto API](http://theopenphotoproject.org/documentation) endpoint layout. For example, the calls in the example above use the following API endpoints:

22
openphoto/__init__.py
View file

@ -1,8 +1,8 @@
from openphoto_http import OpenPhotoHttp
from errors import *
import api_photo
import api_tag
import api_album
from openphoto.openphoto_http import OpenPhotoHttp
from openphoto.errors import *
import openphoto.api_photo
import openphoto.api_tag
import openphoto.api_album
LATEST_API_VERSION = 2
@ -26,9 +26,9 @@ class OpenPhoto(OpenPhotoHttp):
consumer_key, consumer_secret,
token, token_secret, api_version)
self.photos = api_photo.ApiPhotos(self)
self.photo = api_photo.ApiPhoto(self)
self.tags = api_tag.ApiTags(self)
self.tag = api_tag.ApiTag(self)
self.albums = api_album.ApiAlbums(self)
self.album = api_album.ApiAlbum(self)
self.photos = openphoto.api_photo.ApiPhotos(self)
self.photo = openphoto.api_photo.ApiPhoto(self)
self.tags = openphoto.api_tag.ApiTags(self)
self.tag = openphoto.api_tag.ApiTag(self)
self.albums = openphoto.api_album.ApiAlbums(self)
self.album = openphoto.api_album.ApiAlbum(self)

6
openphoto/api_album.py
View file

@ -1,5 +1,4 @@
from errors import *
from objects import Album
from openphoto.objects import Album
class ApiAlbums:
def __init__(self, client):
@ -16,7 +15,8 @@ class ApiAlbum:
def create(self, name, **kwds):
""" Create a new album and return it"""
result = self._client.post("/album/create.json", name=name, **kwds)["result"]
result = self._client.post("/album/create.json",
name=name, **kwds)["result"]
return Album(self._client, result)
def delete(self, album, **kwds):

23
openphoto/api_photo.py
View file

@ -1,7 +1,8 @@
import base64
from errors import *
from objects import Photo
from openphoto.errors import OpenPhotoError
import openphoto.openphoto_http
from openphoto.objects import Photo
class ApiPhotos:
def __init__(self, client):
@ -10,7 +11,7 @@ class ApiPhotos:
def list(self, **kwds):
""" Returns a list of Photo objects """
photos = self._client.get("/photos/list.json", **kwds)["result"]
photos = self._client._result_to_list(photos)
photos = openphoto.openphoto_http.result_to_list(photos)
return [Photo(self._client, photo) for photo in photos]
def update(self, photos, **kwds):
@ -19,7 +20,8 @@ class ApiPhotos:
Returns True if successful.
Raises OpenPhotoError if not.
"""
if not self._client.post("/photos/update.json", ids=photos, **kwds)["result"]:
if not self._client.post("/photos/update.json", ids=photos,
**kwds)["result"]:
raise OpenPhotoError("Update response returned False")
return True
@ -29,7 +31,8 @@ class ApiPhotos:
Returns True if successful.
Raises OpenPhotoError if not.
"""
if not self._client.post("/photos/delete.json", ids=photos, **kwds)["result"]:
if not self._client.post("/photos/delete.json", ids=photos,
**kwds)["result"]:
raise OpenPhotoError("Delete response returned False")
return True
@ -80,13 +83,17 @@ class ApiPhoto:
return photo
def upload(self, photo_file, **kwds):
result = self._client.post("/photo/upload.json", files={'photo': photo_file},
**kwds)["result"]
""" Uploads the specified file to the server """
with open(photo_file, 'rb') as in_file:
result = self._client.post("/photo/upload.json",
files={'photo': in_file},
**kwds)["result"]
return Photo(self._client, result)
def upload_encoded(self, photo_file, **kwds):
""" Base64-encodes and uploads the specified file """
encoded_photo = base64.b64encode(open(photo_file, "rb").read())
with open(photo_file, "rb") as in_file:
encoded_photo = base64.b64encode(in_file.read())
result = self._client.post("/photo/upload.json", photo=encoded_photo,
**kwds)["result"]
return Photo(self._client, result)

8
openphoto/api_tag.py
View file

@ -1,5 +1,4 @@
from errors import *
from objects import Tag
from openphoto.objects import Tag
class ApiTags:
def __init__(self, client):
@ -15,7 +14,10 @@ class ApiTag:
self._client = client
def create(self, tag, **kwds):
""" Create a new tag. The API returns true if the tag was sucessfully created """
"""
Create a new tag.
The API returns true if the tag was sucessfully created
"""
return self._client.post("/tag/create.json", tag=tag, **kwds)["result"]
def delete(self, tag, **kwds):

84
openphoto/config.py Normal file
View file

@ -0,0 +1,84 @@
from __future__ import unicode_literals
import os
try:
from configparser import ConfigParser # Python3
except ImportError:
from ConfigParser import SafeConfigParser as ConfigParser # Python2
try:
import io # Python3
except ImportError:
import StringIO as io # Python2
class Config:
def __init__(self, config_file, host,
consumer_key, consumer_secret,
token, token_secret):
if host is None:
self.config_path = get_config_path(config_file)
config = read_config(self.config_path)
self.host = config['host']
self.consumer_key = config['consumerKey']
self.consumer_secret = config['consumerSecret']
self.token = config['token']
self.token_secret = config['tokenSecret']
else:
self.config_path = None
self.host = host
self.consumer_key = consumer_key
self.consumer_secret = consumer_secret
self.token = token
self.token_secret = token_secret
if host is not None and config_file is not None:
raise ValueError("Cannot specify both host and config_file")
def get_config_path(config_file):
"""
Given the name of a config file, returns the full path
"""
config_path = os.getenv('XDG_CONFIG_HOME')
if not config_path:
config_path = os.path.join(os.getenv('HOME'), ".config")
if not config_file:
config_file = "default"
return os.path.join(config_path, "openphoto", config_file)
def read_config(config_path):
"""
Loads config data from the specified file path.
If config_file doesn't exist, returns an empty authentication config for localhost.
"""
section = "DUMMY"
defaults = {'host': 'localhost',
'consumerKey': '', 'consumerSecret': '',
'token': '', 'tokenSecret':'',
}
# Insert an section header at the start of the config file,
# so ConfigParser can understand it
buf = io.StringIO()
buf.write('[%s]\n' % section)
with io.open(config_path, "r") as conf:
buf.write(conf.read())
buf.seek(0, os.SEEK_SET)
parser = ConfigParser()
parser.optionxform = str # Case-sensitive options
try:
parser.read_file(buf) # Python3
except AttributeError:
parser.readfp(buf) # Python2
# Trim quotes
config = parser.items(section)
config = [(item[0].replace('"', ''), item[1].replace('"', ''))
for item in config]
config = [(item[0].replace("'", ""), item[1].replace("'", ""))
for item in config]
config = dict(config)
# Apply defaults
for key in defaults:
if key not in config:
config[key] = defaults[key]
return config

10
openphoto/errors.py
View file

@ -7,10 +7,8 @@ class OpenPhotoDuplicateError(OpenPhotoError):
pass
class OpenPhoto404Error(Exception):
""" Indicates that an Http 404 error code was received (resource not found) """
"""
Indicates that an Http 404 error code was received
(resource not found)
"""
pass
class NotImplementedError(OpenPhotoError):
""" Indicates that the API function has not yet been coded - please help! """
pass

109
openphoto/main.py
View file

@ -1,38 +1,51 @@
#!/usr/bin/env python
import os
import sys
import string
import urllib
import json
from optparse import OptionParser
try:
import json
except ImportError:
import simplejson as json
from openphoto import OpenPhoto
CONFIG_ERROR = """
You must create a configuration file with the following contents:
host = your.host.com
consumerKey = your_consumer_key
consumerSecret = your_consumer_secret
token = your_access_token
tokenSecret = your_access_token_secret
To get your credentials:
* Log into your Trovebox site
* Click the arrow on the top-right and select 'Settings'.
* Click the 'Create a new app' button.
* Click the 'View' link beside the newly created app.
"""
#################################################################
def main(args=sys.argv[1:]):
usage = "%prog --help"
parser = OptionParser(usage, add_help_option=False)
parser.add_option('-c', '--config', action='store', type='string', dest='config_file',
help="Configuration file to use")
parser.add_option('-h', '-H', '--host', action='store', type='string', dest='host',
help="Hostname of the OpenPhoto server (overrides config_file)")
parser.add_option('-X', action='store', type='choice', dest='method', choices=('GET', 'POST'),
help="Method to use (GET or POST)", default="GET")
parser.add_option('-F', action='append', type='string', dest='fields',
help="Fields")
parser.add_option('-e', action='store', type='string', dest='endpoint',
default='/photos/list.json',
help="Endpoint to call")
parser.add_option('-p', action="store_true", dest="pretty", default=False,
help="Pretty print the json")
parser.add_option('-v', action="store_true", dest="verbose", default=False,
help="Verbose output")
parser.add_option('--help', action="store_true", help='show this help message')
parser.add_option('-c', '--config', help="Configuration file to use",
action='store', type='string', dest='config_file')
parser.add_option('-h', '-H', '--host',
help=("Hostname of the OpenPhoto server "
"(overrides config_file)"),
action='store', type='string', dest='host')
parser.add_option('-X', help="Method to use (GET or POST)",
action='store', type='choice', dest='method',
choices=('GET', 'POST'), default="GET")
parser.add_option('-F', help="Endpoint field",
action='append', type='string', dest='fields')
parser.add_option('-e', help="Endpoint to call",
action='store', type='string', dest='endpoint',
default='/photos/list.json')
parser.add_option('-p', help="Pretty print the json",
action="store_true", dest="pretty", default=False)
parser.add_option('-v', help="Verbose output",
action="store_true", dest="verbose", default=False)
parser.add_option('--help', help='show this help message',
action="store_true")
options, args = parser.parse_args(args)
@ -46,7 +59,7 @@ def main(args=sys.argv[1:]):
params = {}
if options.fields:
for field in options.fields:
(key, value) = string.split(field, '=')
(key, value) = field.split('=')
params[key] = value
# Host option overrides config file settings
@ -56,42 +69,33 @@ def main(args=sys.argv[1:]):
try:
client = OpenPhoto(config_file=options.config_file)
except IOError as error:
print error
print
print "You must create a configuration file with the following contents:"
print " host = your.host.com"
print " consumerKey = your_consumer_key"
print " consumerSecret = your_consumer_secret"
print " token = your_access_token"
print " tokenSecret = your_access_token_secret"
print
print "To get your credentials:"
print " * Log into your Trovebox site"
print " * Click the arrow on the top-right and select 'Settings'."
print " * Click the 'Create a new app' button."
print " * Click the 'View' link beside the newly created app."
print
print error
print(error)
print(CONFIG_ERROR)
print(error)
sys.exit(1)
if options.method == "GET":
result = client.get(options.endpoint, process_response=False, **params)
result = client.get(options.endpoint, process_response=False,
**params)
else:
params, files = extract_files(params)
result = client.post(options.endpoint, process_response=False, files=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, config['host'], options.endpoint)
if len( params ) > 0:
print "Fields:"
for kv in params.iteritems():
print " %s=%s" % kv
print "==========\n"
print("==========\nMethod: %s\nHost: %s\nEndpoint: %s" %
(options.method, client.host, options.endpoint))
if params:
print("Fields:")
for key, value in params.items():
print(" %s=%s" % (key, value))
print("==========\n")
if options.pretty:
print json.dumps(json.loads(result), sort_keys=True, indent=4, separators=(',',':'))
print(json.dumps(json.loads(result), sort_keys=True,
indent=4, separators=(',',':')))
else:
print result
print(result)
def extract_files(params):
"""
@ -105,8 +109,9 @@ def extract_files(params):
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:]
if (name == "photo" and params[name].startswith("@") and
os.path.isfile(os.path.expanduser(params[name][1:]))):
files[name] = open(params[name][1:], 'rb')
else:
updated_params[name] = params[name]

31
openphoto/multipart_post.py
View file

@ -1,31 +0,0 @@
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

63
openphoto/objects.py
View file

@ -1,9 +1,13 @@
import urllib
from errors import *
try:
from urllib.parse import quote # Python3
except ImportError:
from urllib import quote # Python2
class OpenPhotoObject:
""" Base object supporting the storage of custom fields as attributes """
def __init__(self, openphoto, json_dict):
self.id = None
self.name = None
self._openphoto = openphoto
self._json_dict = json_dict
self._set_fields(json_dict)
@ -26,9 +30,9 @@ class OpenPhotoObject:
self._set_fields(json_dict)
def __repr__(self):
if hasattr(self, "name"):
if self.name is not None:
return "<%s name='%s'>" % (self.__class__, self.name)
elif hasattr(self, "id"):
elif self.id is not None:
return "<%s id='%s'>" % (self.__class__, self.id)
else:
return "<%s>" % (self.__class__)
@ -45,14 +49,15 @@ class Photo(OpenPhotoObject):
Returns True if successful.
Raises an OpenPhotoError if not.
"""
result = self._openphoto.post("/photo/%s/delete.json" % self.id, **kwds)["result"]
result = self._openphoto.post("/photo/%s/delete.json" %
self.id, **kwds)["result"]
self._replace_fields({})
return result
def edit(self, **kwds):
""" Returns an HTML form to edit the photo """
result = self._openphoto.get("/photo/%s/edit.json" % self.id,
**kwds)["result"]
result = self._openphoto.get("/photo/%s/edit.json" %
self.id, **kwds)["result"]
return result["markup"]
def replace(self, photo_file, **kwds):
@ -63,8 +68,8 @@ class Photo(OpenPhotoObject):
def update(self, **kwds):
""" Update this photo with the specified parameters """
new_dict = self._openphoto.post("/photo/%s/update.json" % self.id,
**kwds)["result"]
new_dict = self._openphoto.post("/photo/%s/update.json" %
self.id, **kwds)["result"]
self._replace_fields(new_dict)
def view(self, **kwds):
@ -72,8 +77,8 @@ class Photo(OpenPhotoObject):
Used to view the photo at a particular size.
Updates the photo's fields with the response.
"""
new_dict = self._openphoto.get("/photo/%s/view.json" % self.id,
**kwds)["result"]
new_dict = self._openphoto.get("/photo/%s/view.json" %
self.id, **kwds)["result"]
self._replace_fields(new_dict)
def dynamic_url(self, **kwds):
@ -84,8 +89,8 @@ class Photo(OpenPhotoObject):
Returns a dict containing the next and previous photo lists
(there may be more than one next/previous photo returned).
"""
result = self._openphoto.get("/photo/%s/nextprevious.json" % self.id,
**kwds)["result"]
result = self._openphoto.get("/photo/%s/nextprevious.json" %
self.id, **kwds)["result"]
value = {}
if "next" in result:
# Workaround for APIv1
@ -112,12 +117,13 @@ class Photo(OpenPhotoObject):
Performs transformation specified in **kwds
Example: transform(rotate=90)
"""
new_dict = self._openphoto.post("/photo/%s/transform.json" % self.id,
**kwds)["result"]
new_dict = self._openphoto.post("/photo/%s/transform.json" %
self.id, **kwds)["result"]
# APIv1 doesn't return the transformed photo (frontend issue #955)
if isinstance(new_dict, bool):
new_dict = self._openphoto.get("/photo/%s/view.json" % self.id)["result"]
new_dict = self._openphoto.get("/photo/%s/view.json" %
self.id)["result"]
self._replace_fields(new_dict)
@ -128,13 +134,14 @@ class Tag(OpenPhotoObject):
Returns True if successful.
Raises an OpenPhotoError if not.
"""
result = self._openphoto.post("/tag/%s/delete.json" % urllib.quote(self.id), **kwds)["result"]
result = self._openphoto.post("/tag/%s/delete.json" %
quote(self.id), **kwds)["result"]
self._replace_fields({})
return result
def update(self, **kwds):
""" Update this tag with the specified parameters """
new_dict = self._openphoto.post("/tag/%s/update.json" % urllib.quote(self.id),
new_dict = self._openphoto.post("/tag/%s/update.json" % quote(self.id),
**kwds)["result"]
self._replace_fields(new_dict)
@ -142,15 +149,17 @@ class Tag(OpenPhotoObject):
class Album(OpenPhotoObject):
def __init__(self, openphoto, json_dict):
OpenPhotoObject.__init__(self, openphoto, json_dict)
self.photos = None
self.cover = None
self._update_fields_with_objects()
def _update_fields_with_objects(self):
""" Convert dict fields into objects, where appropriate """
# Update the cover with a photo object
if hasattr(self, "cover") and isinstance(self.cover, dict):
if isinstance(self.cover, dict):
self.cover = Photo(self._openphoto, self.cover)
# Update the photo list with photo objects
if hasattr(self, "photos") and isinstance(self.photos, list):
if isinstance(self.photos, list):
for i, photo in enumerate(self.photos):
if isinstance(photo, dict):
self.photos[i] = Photo(self._openphoto, photo)
@ -161,7 +170,8 @@ class Album(OpenPhotoObject):
Returns True if successful.
Raises an OpenPhotoError if not.
"""
result = self._openphoto.post("/album/%s/delete.json" % self.id, **kwds)["result"]
result = self._openphoto.post("/album/%s/delete.json" %
self.id, **kwds)["result"]
self._replace_fields({})
return result
@ -176,12 +186,13 @@ class Album(OpenPhotoObject):
def update(self, **kwds):
""" Update this album with the specified parameters """
new_dict = self._openphoto.post("/album/%s/update.json" % self.id,
**kwds)["result"]
new_dict = self._openphoto.post("/album/%s/update.json" %
self.id, **kwds)["result"]
# APIv1 doesn't return the updated album (frontend issue #937)
if isinstance(new_dict, bool):
new_dict = self._openphoto.get("/album/%s/view.json" % self.id)["result"]
new_dict = self._openphoto.get("/album/%s/view.json" %
self.id)["result"]
self._replace_fields(new_dict)
self._update_fields_with_objects()
@ -191,7 +202,7 @@ class Album(OpenPhotoObject):
Requests the full contents of the album.
Updates the album's fields with the response.
"""
result = self._openphoto.get("/album/%s/view.json" % self.id,
**kwds)["result"]
result = self._openphoto.get("/album/%s/view.json" %
self.id, **kwds)["result"]
self._replace_fields(result)
self._update_fields_with_objects()

217
openphoto/openphoto_http.py
View file

@ -1,19 +1,25 @@
import os
import oauth2 as oauth
import urlparse
import urllib
import httplib2
from __future__ import unicode_literals
import sys
import requests
import requests_oauthlib
import logging
import StringIO
import ConfigParser
try:
import json
from urllib.parse import urlunparse # Python3
except ImportError:
import simplejson as json
from urlparse import urlunparse # Python2
from objects import OpenPhotoObject
from errors import *
from multipart_post import encode_multipart_formdata
from openphoto.objects import OpenPhotoObject
from openphoto.errors import *
from openphoto.config import Config
if sys.version < '3':
TEXT_TYPE = unicode
# requests_oauth needs to decode to ascii for Python2
OAUTH_DECODING = "utf-8"
else:
TEXT_TYPE = str
# requests_oauth needs to use (unicode) strings for Python3
OAUTH_DECODING = None
DUPLICATE_RESPONSE = {"code": 409,
"message": "This photo already exists"}
@ -37,23 +43,11 @@ class OpenPhotoHttp:
self._logger = logging.getLogger("openphoto")
if host is None:
self.config_path = self._get_config_path(config_file)
config = self._read_config(self.config_path)
self._host = config['host']
self._consumer_key = config['consumerKey']
self._consumer_secret = config['consumerSecret']
self._token = config['token']
self._token_secret = config['tokenSecret']
else:
self._host = host
self._consumer_key = consumer_key
self._consumer_secret = consumer_secret
self._token = token
self._token_secret = token_secret
self.config = Config(config_file, host,
consumer_key, consumer_secret,
token, token_secret)
if host is not None and config_file is not None:
raise ValueError("Cannot specify both host and config_file")
self.host = self.config.host
# Remember the most recent HTTP request and response
self.last_url = None
@ -76,32 +70,35 @@ class OpenPhotoHttp:
endpoint = "/" + endpoint
if self._api_version is not None:
endpoint = "/v%d%s" % (self._api_version, endpoint)
url = urlparse.urlunparse(('http', self._host, endpoint, '',
urllib.urlencode(params), ''))
if self._consumer_key:
consumer = oauth.Consumer(self._consumer_key, self._consumer_secret)
token = oauth.Token(self._token, self._token_secret)
client = oauth.Client(consumer, token)
else:
client = httplib2.Http()
url = urlunparse(('http', self.host, endpoint, '', '', ''))
response, content = client.request(url, "GET")
if self.config.consumer_key:
auth = requests_oauthlib.OAuth1(self.config.consumer_key,
self.config.consumer_secret,
self.config.token,
self.config.token_secret,
decoding=OAUTH_DECODING)
else:
auth = None
with requests.Session() as session:
response = session.get(url, params=params, auth=auth)
self._logger.info("============================")
self._logger.info("GET %s" % url)
self._logger.info("---")
self._logger.info(content)
self._logger.info(response.text)
self.last_url = url
self.last_params = params
self.last_response = (response, content)
self.last_response = response
if process_response:
return self._process_response(response, content)
return self._process_response(response)
else:
return content
return response.text
def post(self, endpoint, process_response=True, files = {}, **params):
def post(self, endpoint, process_response=True, files=None, **params):
"""
Performs an HTTP POST to the specified endpoint (API path),
passing parameters if given.
@ -117,24 +114,26 @@ class OpenPhotoHttp:
endpoint = "/" + endpoint
if self._api_version is not None:
endpoint = "/v%d%s" % (self._api_version, endpoint)
url = urlparse.urlunparse(('http', self._host, endpoint, '', '', ''))
url = urlunparse(('http', self.host, endpoint, '', '', ''))
if not self._consumer_key:
if not self.config.consumer_key:
raise OpenPhotoError("Cannot issue POST without OAuth tokens")
consumer = oauth.Consumer(self._consumer_key, self._consumer_secret)
token = oauth.Token(self._token, self._token_secret)
client = oauth.Client(consumer, token)
if files:
# Parameters must be signed and encoded into the multipart body
signed_params = self._sign_params(client, url, params)
headers, body = encode_multipart_formdata(signed_params, files)
else:
body = urllib.urlencode(params)
headers = None
response, content = client.request(url, "POST", body, headers)
auth = requests_oauthlib.OAuth1(self.config.consumer_key,
self.config.consumer_secret,
self.config.token,
self.config.token_secret,
decoding=OAUTH_DECODING)
with requests.Session() as session:
if files:
# Need to pass parameters as URL query, so they get OAuth signed
response = session.post(url, params=params,
files=files, auth=auth)
else:
# Passing parameters as URL query doesn't work
# if there are no files to send.
# Send them as form data instead.
response = session.post(url, data=params, auth=auth)
self._logger.info("============================")
self._logger.info("POST %s" % url)
@ -142,27 +141,16 @@ class OpenPhotoHttp:
if files:
self._logger.info("files: %s" % repr(files))
self._logger.info("---")
self._logger.info(content)
self._logger.info(response.text)
self.last_url = url
self.last_params = params
self.last_response = (response, content)
self.last_response = response
if process_response:
return self._process_response(response, content)
return self._process_response(response)
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()))
return response.text
@staticmethod
def _process_params(params):
@ -173,9 +161,9 @@ class OpenPhotoHttp:
if isinstance(value, OpenPhotoObject):
value = value.id
# Use UTF-8 encoding
if isinstance(value, unicode):
value = value.encode('utf-8')
# Ensure value is UTF-8 encoded
if isinstance(value, TEXT_TYPE):
value = value.encode("utf-8")
# Handle lists
if isinstance(value, list):
@ -185,8 +173,8 @@ class OpenPhotoHttp:
for i, item in enumerate(new_list):
if isinstance(item, OpenPhotoObject):
new_list[i] = item.id
# Convert list to unicode string
value = u','.join([unicode(item) for item in new_list])
# Convert list to string
value = ','.join([str(item) for item in new_list])
# Handle booleans
if isinstance(value, bool):
@ -196,24 +184,26 @@ class OpenPhotoHttp:
return processed_params
@staticmethod
def _process_response(response, content):
def _process_response(response):
"""
Decodes the JSON response, returning a dict.
Raises an exception if an invalid response code is received.
"""
try:
json_response = json.loads(content)
json_response = response.json()
code = json_response["code"]
message = json_response["message"]
except ValueError, KeyError:
except (ValueError, KeyError):
# Response wasn't OpenPhoto JSON - check the HTTP status code
if 200 <= response.status < 300:
if 200 <= response.status_code < 300:
# Status code was valid, so just reraise the exception
raise
elif response.status == 404:
raise OpenPhoto404Error("HTTP Error %d: %s" % (response.status, response.reason))
elif response.status_code == 404:
raise OpenPhoto404Error("HTTP Error %d: %s" %
(response.status_code, response.reason))
else:
raise OpenPhotoError("HTTP Error %d: %s" % (response.status, response.reason))
raise OpenPhotoError("HTTP Error %d: %s" %
(response.status_code, response.reason))
if 200 <= code < 300:
return json_response
@ -223,54 +213,11 @@ class OpenPhotoHttp:
else:
raise OpenPhotoError("Code %d: %s" % (code, message))
@staticmethod
def _result_to_list(result):
""" Handle the case where the result contains no items """
if not result:
return []
if result[0]["totalRows"] == 0:
return []
else:
return result
@staticmethod
def _get_config_path(config_file):
config_path = os.getenv('XDG_CONFIG_HOME')
if not config_path:
config_path = os.path.join(os.getenv('HOME'), ".config")
if not config_file:
config_file = "default"
return os.path.join(config_path, "openphoto", config_file)
def _read_config(self, config_file):
"""
Loads config data from the specified file.
If config_file doesn't exist, returns an empty authentication config for localhost.
"""
section = "DUMMY"
defaults = {'host': 'localhost',
'consumerKey': '', 'consumerSecret': '',
'token': '', 'tokenSecret':'',
}
# Insert an section header at the start of the config file, so ConfigParser can understand it
buf = StringIO.StringIO()
buf.write('[%s]\n' % section)
buf.write(open(config_file).read())
buf.seek(0, os.SEEK_SET)
parser = ConfigParser.SafeConfigParser()
parser.optionxform = str # Case-sensitive options
parser.readfp(buf)
# Trim quotes
config = parser.items(section)
config = [(item[0].replace('"', ''), item[1].replace('"', '')) for item in config]
config = [(item[0].replace("'", ""), item[1].replace("'", "")) for item in config]
config = dict(config)
# Apply defaults
for key in defaults:
if key not in config:
config[key] = defaults[key]
return config
def result_to_list(result):
""" Handle the case where the result contains no items """
if not result:
return []
if result[0]["totalRows"] == 0:
return []
else:
return result

8
setup.py
View file

@ -1,10 +1,6 @@
#!/usr/bin/env python
requires = ['oauth2', 'httplib2']
try:
import json
except ImportError:
requires.append('simplejson')
requires = ['requests', 'requests-oauthlib']
try:
from setuptools import setup
@ -19,7 +15,7 @@ except ImportError:
'requires': requires}
setup(name='openphoto',
version='0.2',
version='0.3',
description='Client library for the openphoto project',
author='James Walker',
author_email='walkah@walkah.net',

4
tests/api_versions/test_v1.py
View file

@ -1,7 +1,3 @@
try:
import unittest2 as unittest
except ImportError:
import unittest
from tests import test_albums, test_photos, test_tags
class TestAlbumsV1(test_albums.TestAlbums):

9
tests/api_versions/test_v2.py
View file

@ -4,14 +4,17 @@ except ImportError:
import unittest
from tests import test_base, test_albums, test_photos, test_tags
@unittest.skipIf(test_base.get_test_server_api() < 2, "Don't test future API versions")
@unittest.skipIf(test_base.get_test_server_api() < 2,
"Don't test future API versions")
class TestAlbumsV2(test_albums.TestAlbums):
api_version = 2
@unittest.skipIf(test_base.get_test_server_api() < 2, "Don't test future API versions")
@unittest.skipIf(test_base.get_test_server_api() < 2,
"Don't test future API versions")
class TestPhotosV2(test_photos.TestPhotos):
api_version = 2
@unittest.skipIf(test_base.get_test_server_api() < 2, "Don't test future API versions")
@unittest.skipIf(test_base.get_test_server_api() < 2,
"Don't test future API versions")
class TestTagsV2(test_tags.TestTags):
api_version = 2

28
tests/test_albums.py
View file

@ -1,11 +1,6 @@
try:
import unittest2 as unittest
except ImportError:
import unittest
import openphoto
import test_base
import tests.test_base
class TestAlbums(test_base.TestBase):
class TestAlbums(tests.test_base.TestBase):
testcase_name = "album API"
def test_create_delete(self):
@ -16,22 +11,26 @@ class TestAlbums(test_base.TestBase):
# Check the return value
self.assertEqual(album.name, album_name)
# Check that the album now exists
self.assertIn(album_name, [a.name for a in self.client.albums.list()])
self.assertIn(album_name,
[a.name for a in self.client.albums.list()])
# Delete the album
self.assertTrue(self.client.album.delete(album.id))
# Check that the album is now gone
self.assertNotIn(album_name, [a.name for a in self.client.albums.list()])
self.assertNotIn(album_name,
[a.name for a in self.client.albums.list()])
# Create it again, and delete it using the Album object
album = self.client.album.create(album_name)
self.assertTrue(album.delete())
# Check that the album is now gone
self.assertNotIn(album_name, [a.name for a in self.client.albums.list()])
self.assertNotIn(album_name,
[a.name for a in self.client.albums.list()])
def test_update(self):
""" Test that an album can be updated """
# Update the album using the OpenPhoto class, passing in the album object
# Update the album using the OpenPhoto class,
# passing in the album object
new_name = "New Name"
self.client.album.update(self.albums[0], name=new_name)
@ -57,7 +56,6 @@ class TestAlbums(test_base.TestBase):
def test_view(self):
""" Test the album view """
album = self.albums[0]
self.assertFalse(hasattr(album, "photos"))
# Get the photos in the album using the Album object directly
album.view(includeElements=True)
@ -67,15 +65,15 @@ class TestAlbums(test_base.TestBase):
def test_form(self):
""" If album.form gets implemented, write a test! """
with self.assertRaises(openphoto.NotImplementedError):
with self.assertRaises(NotImplementedError):
self.client.album.form(None)
def test_add_photos(self):
""" If album.add_photos gets implemented, write a test! """
with self.assertRaises(openphoto.NotImplementedError):
with self.assertRaises(NotImplementedError):
self.client.album.add_photos(None, None)
def test_remove_photos(self):
""" If album.remove_photos gets implemented, write a test! """
with self.assertRaises(openphoto.NotImplementedError):
with self.assertRaises(NotImplementedError):
self.client.album.remove_photos(None, None)

45
tests/test_base.py
View file

@ -1,14 +1,17 @@
from __future__ import print_function
import sys
import os
import logging
try:
import unittest2 as unittest
import unittest2 as unittest # Python2.6
except ImportError:
import unittest
import logging
import openphoto
def get_test_server_api():
return int(os.getenv("OPENPHOTO_TEST_SERVER_API", openphoto.LATEST_API_VERSION))
return int(os.getenv("OPENPHOTO_TEST_SERVER_API",
openphoto.LATEST_API_VERSION))
class TestBase(unittest.TestCase):
TEST_TITLE = "Test Image - delete me!"
@ -22,7 +25,7 @@ class TestBase(unittest.TestCase):
debug = (os.getenv("OPENPHOTO_TEST_DEBUG", "0") == "1")
def __init__(self, *args, **kwds):
unittest.TestCase.__init__(self, *args, **kwds)
super(TestBase, self).__init__(*args, **kwds)
self.photos = []
logging.basicConfig(filename="tests.log",
@ -35,9 +38,9 @@ class TestBase(unittest.TestCase):
""" Ensure there is nothing on the server before running any tests """
if cls.debug:
if cls.api_version is None:
print "\nTesting Latest %s" % cls.testcase_name
print("\nTesting Latest %s" % cls.testcase_name)
else:
print "\nTesting %s v%d" % (cls.testcase_name, cls.api_version)
print("\nTesting %s v%d" % (cls.testcase_name, cls.api_version))
cls.client = openphoto.OpenPhoto(config_file=cls.config_file,
api_version=cls.api_version)
@ -45,17 +48,17 @@ class TestBase(unittest.TestCase):
if cls.client.photos.list() != []:
raise ValueError("The test server (%s) contains photos. "
"Please delete them before running the tests"
% cls.client._host)
% cls.client.host)
if cls.client.tags.list() != []:
raise ValueError("The test server (%s) contains tags. "
"Please delete them before running the tests"
% cls.client._host)
% cls.client.host)
if cls.client.albums.list() != []:
raise ValueError("The test server (%s) contains albums. "
"Please delete them before running the tests"
% cls.client._host)
% cls.client.host)
@classmethod
def tearDownClass(cls):
@ -71,9 +74,9 @@ class TestBase(unittest.TestCase):
self.photos = self.client.photos.list()
if len(self.photos) != 3:
if self.debug:
print "[Regenerating Photos]"
print("[Regenerating Photos]")
else:
print " ",
print(" ", end='')
sys.stdout.flush()
if len(self.photos) > 0:
self._delete_all()
@ -85,16 +88,16 @@ class TestBase(unittest.TestCase):
self.tags[0].id != self.TEST_TAG or
str(self.tags[0].count) != "3"):
if self.debug:
print "[Regenerating Tags]"
print("[Regenerating Tags]")
else:
print " ",
print(" ", end='')
sys.stdout.flush()
self._delete_all()
self._create_test_photos()
self.photos = self.client.photos.list()
self.tags = self.client.tags.list()
if len(self.tags) != 1:
print "Tags: %s" % self.tags
print("Tags: %s" % self.tags)
raise Exception("Tag creation failed")
self.albums = self.client.albums.list()
@ -102,9 +105,9 @@ class TestBase(unittest.TestCase):
self.albums[0].name != self.TEST_ALBUM or
self.albums[0].count != "3"):
if self.debug:
print "[Regenerating Albums]"
print("[Regenerating Albums]")
else:
print " ",
print(" ", end='')
sys.stdout.flush()
self._delete_all()
self._create_test_photos()
@ -112,13 +115,13 @@ class TestBase(unittest.TestCase):
self.tags = self.client.tags.list()
self.albums = self.client.albums.list()
if len(self.albums) != 1:
print "Albums: %s" % self.albums
print("Albums: %s" % self.albums)
raise Exception("Album creation failed")
logging.info("\nRunning %s..." % self.id())
logging.info("\nRunning %s...", self.id())
def tearDown(self):
logging.info("Finished %s\n" % self.id())
logging.info("Finished %s\n", self.id())
@classmethod
def _create_test_photos(cls):
@ -141,9 +144,11 @@ class TestBase(unittest.TestCase):
@classmethod
def _delete_all(cls):
""" Remove all photos, tags and albums """
photos = cls.client.photos.list()
if len(photos) > cls.MAXIMUM_TEST_PHOTOS:
raise ValueError("There too many photos on the test server - must always be less than %d."
raise ValueError("There too many photos on the test server "
"- must always be less than %d."
% cls.MAXIMUM_TEST_PHOTOS)
for photo in photos:
photo.delete()

90
tests/test_config.py
View file

@ -1,10 +1,11 @@
try:
import unittest2 as unittest
except ImportError:
import unittest
import os
import shutil
import openphoto
try:
import unittest2 as unittest # Python2.6
except ImportError:
import unittest
from openphoto import OpenPhoto
CONFIG_HOME_PATH = os.path.join("tests", "config")
CONFIG_PATH = os.path.join(CONFIG_HOME_PATH, "openphoto")
@ -26,68 +27,73 @@ class TestConfig(unittest.TestCase):
os.environ["XDG_CONFIG_HOME"] = self.original_xdg_config_home
shutil.rmtree(CONFIG_HOME_PATH, ignore_errors=True)
def create_config(self, config_file, host):
f = open(os.path.join(CONFIG_PATH, config_file), "w")
f.write("host = %s\n" % host)
f.write("# Comment\n\n")
f.write("consumerKey = \"%s_consumer_key\"\n" % config_file)
f.write("\"consumerSecret\" = %s_consumer_secret\n" % config_file)
f.write("'token'=%s_token\n" % config_file)
f.write("tokenSecret = '%s_token_secret'\n" % config_file)
@staticmethod
def create_config(config_file, host):
with open(os.path.join(CONFIG_PATH, config_file), "w") as conf:
conf.write("host = %s\n" % host)
conf.write("# Comment\n\n")
conf.write("consumerKey = \"%s_consumer_key\"\n" % config_file)
conf.write("\"consumerSecret\" = %s_consumer_secret\n" % config_file)
conf.write("'token'=%s_token\n" % config_file)
conf.write("tokenSecret = '%s_token_secret'\n" % config_file)
def test_default_config(self):
""" Ensure the default config is loaded """
self.create_config("default", "Test Default Host")
client = openphoto.OpenPhoto()
self.assertEqual(client._host, "Test Default Host")
self.assertEqual(client._consumer_key, "default_consumer_key")
self.assertEqual(client._consumer_secret, "default_consumer_secret")
self.assertEqual(client._token, "default_token")
self.assertEqual(client._token_secret, "default_token_secret")
client = OpenPhoto()
config = client.config
self.assertEqual(client.host, "Test Default Host")
self.assertEqual(config.consumer_key, "default_consumer_key")
self.assertEqual(config.consumer_secret, "default_consumer_secret")
self.assertEqual(config.token, "default_token")
self.assertEqual(config.token_secret, "default_token_secret")
def test_custom_config(self):
""" Ensure a custom config can be loaded """
self.create_config("default", "Test Default Host")
self.create_config("custom", "Test Custom Host")
client = openphoto.OpenPhoto(config_file="custom")
self.assertEqual(client._host, "Test Custom Host")
self.assertEqual(client._consumer_key, "custom_consumer_key")
self.assertEqual(client._consumer_secret, "custom_consumer_secret")
self.assertEqual(client._token, "custom_token")
self.assertEqual(client._token_secret, "custom_token_secret")
client = OpenPhoto(config_file="custom")
config = client.config
self.assertEqual(client.host, "Test Custom Host")
self.assertEqual(config.consumer_key, "custom_consumer_key")
self.assertEqual(config.consumer_secret, "custom_consumer_secret")
self.assertEqual(config.token, "custom_token")
self.assertEqual(config.token_secret, "custom_token_secret")
def test_full_config_path(self):
""" Ensure a full custom config path can be loaded """
self.create_config("path", "Test Path Host")
full_path = os.path.abspath(CONFIG_PATH)
client = openphoto.OpenPhoto(config_file=os.path.join(full_path, "path"))
self.assertEqual(client._host, "Test Path Host")
self.assertEqual(client._consumer_key, "path_consumer_key")
self.assertEqual(client._consumer_secret, "path_consumer_secret")
self.assertEqual(client._token, "path_token")
self.assertEqual(client._token_secret, "path_token_secret")
client = OpenPhoto(config_file=os.path.join(full_path, "path"))
config = client.config
self.assertEqual(client.host, "Test Path Host")
self.assertEqual(config.consumer_key, "path_consumer_key")
self.assertEqual(config.consumer_secret, "path_consumer_secret")
self.assertEqual(config.token, "path_token")
self.assertEqual(config.token_secret, "path_token_secret")
def test_host_override(self):
""" Ensure that specifying a host overrides the default config """
self.create_config("default", "Test Default Host")
client = openphoto.OpenPhoto(host="host_override")
self.assertEqual(client._host, "host_override")
self.assertEqual(client._consumer_key, "")
self.assertEqual(client._consumer_secret, "")
self.assertEqual(client._token, "")
self.assertEqual(client._token_secret, "")
client = OpenPhoto(host="host_override")
config = client.config
self.assertEqual(config.host, "host_override")
self.assertEqual(config.consumer_key, "")
self.assertEqual(config.consumer_secret, "")
self.assertEqual(config.token, "")
self.assertEqual(config.token_secret, "")
def test_missing_config_files_raise_exceptions(self):
def test_missing_config_files(self):
""" Ensure that missing config files raise exceptions """
with self.assertRaises(IOError):
openphoto.OpenPhoto()
OpenPhoto()
with self.assertRaises(IOError):
openphoto.OpenPhoto(config_file="custom")
OpenPhoto(config_file="custom")
def test_host_and_config_file_raises_exception(self):
def test_host_and_config_file(self):
""" It's not valid to specify both a host and a config_file """
self.create_config("custom", "Test Custom Host")
with self.assertRaises(ValueError):
openphoto.OpenPhoto(config_file="custom", host="host_override")
OpenPhoto(config_file="custom", host="host_override")

46
tests/test_framework.py
View file

@ -1,37 +1,45 @@
try:
import unittest2 as unittest
except ImportError:
import unittest
import logging
import openphoto
import test_base
class TestFramework(test_base.TestBase):
import openphoto
import tests.test_base
class TestFramework(tests.test_base.TestBase):
testcase_name = "framework"
def setUp(self):
"""Override the default setUp, since we don't need a populated database"""
logging.info("\nRunning %s..." % self.id())
"""
Override the default setUp, since we don't need a populated database
"""
logging.info("\nRunning %s...", self.id())
def test_api_version_zero(self):
# API v0 has a special hello world message
"""
API v0 has a special hello world message
"""
client = openphoto.OpenPhoto(config_file=self.config_file,
api_version=0)
result = client.get("hello.json")
self.assertEqual(result['message'], "Hello, world! This is version zero of the API!")
self.assertEqual(result['message'],
"Hello, world! This is version zero of the API!")
self.assertEqual(result['result']['__route__'], "/v0/hello.json")
def test_specified_api_version(self):
# For all API versions >0, we get a generic hello world message
for api_version in range(1, test_base.get_test_server_api() + 1):
"""
For all API versions >0, we get a generic hello world message
"""
for api_version in range(1, tests.test_base.get_test_server_api() + 1):
client = openphoto.OpenPhoto(config_file=self.config_file,
api_version=api_version)
result = client.get("hello.json")
self.assertEqual(result['message'], "Hello, world!")
self.assertEqual(result['result']['__route__'], "/v%d/hello.json" % api_version)
self.assertEqual(result['result']['__route__'],
"/v%d/hello.json" % api_version)
def test_unspecified_api_version(self):
# If the API version is unspecified, we get a generic hello world message
"""
If the API version is unspecified,
we get a generic hello world message.
"""
client = openphoto.OpenPhoto(config_file=self.config_file,
api_version=None)
result = client.get("hello.json")
@ -39,9 +47,11 @@ class TestFramework(test_base.TestBase):
self.assertEqual(result['result']['__route__'], "/hello.json")
def test_future_api_version(self):
# If the API version is unsupported, we should get an error
# (it's a ValueError, since the returned 404 HTML page is not valid JSON)
"""
If the API version is unsupported, we should get an error
(ValueError, since the returned 404 HTML page is not valid JSON)
"""
client = openphoto.OpenPhoto(config_file=self.config_file,
api_version=openphoto.LATEST_API_VERSION + 1)
api_version=openphoto.LATEST_API_VERSION + 1)
with self.assertRaises(openphoto.OpenPhoto404Error):
client.get("hello.json")

20
tests/test_photos.py
View file

@ -1,11 +1,9 @@
try:
import unittest2 as unittest
except ImportError:
import unittest
import openphoto
import test_base
from __future__ import unicode_literals
class TestPhotos(test_base.TestBase):
import openphoto
import tests.test_base
class TestPhotos(tests.test_base.TestBase):
testcase_name = "photo API"
def test_delete_upload(self):
@ -72,7 +70,7 @@ class TestPhotos(test_base.TestBase):
def test_update(self):
""" Update a photo by editing the title """
title = u"\xfcmlaut" # umlauted umlaut
title = "\xfcmlaut" # umlauted umlaut
# Get a photo and check that it doesn't have the magic title
photo = self.photos[0]
self.assertNotEqual(photo.title, title)
@ -140,17 +138,17 @@ class TestPhotos(test_base.TestBase):
def test_replace(self):
""" If photo.replace gets implemented, write a test! """
with self.assertRaises(openphoto.NotImplementedError):
with self.assertRaises(NotImplementedError):
self.client.photo.replace(None, None)
def test_replace_encoded(self):
""" If photo.replace_encoded gets implemented, write a test! """
with self.assertRaises(openphoto.NotImplementedError):
with self.assertRaises(NotImplementedError):
self.client.photo.replace_encoded(None, None)
def test_dynamic_url(self):
""" If photo.dynamic_url gets implemented, write a test! """
with self.assertRaises(openphoto.NotImplementedError):
with self.assertRaises(NotImplementedError):
self.client.photo.dynamic_url(None)
def test_transform(self):

17
tests/test_tags.py
View file

@ -1,13 +1,13 @@
try:
import unittest2 as unittest
import unittest2 as unittest # Python2.6
except ImportError:
import unittest
import openphoto
import test_base
@unittest.skipIf(test_base.get_test_server_api() == 1,
import tests.test_base
@unittest.skipIf(tests.test_base.get_test_server_api() == 1,
"The tag API didn't work at v1 - see frontend issue #927")
class TestTags(test_base.TestBase):
class TestTags(tests.test_base.TestBase):
testcase_name = "tag API"
def test_create_delete(self, tag_id="create_tag"):
@ -46,9 +46,10 @@ class TestTags(test_base.TestBase):
# Also remove the tag from the photo
self.photos[0].update(tagsRemove=tag_id)
# TODO: Un-skip and update this tests once there are tag fields that can be updated.
# The owner field cannot be updated.
@unittest.skip("Can't test the tag.update endpoint, since there are no fields that can be updated")
# TODO: Un-skip and update this tests once there are tag fields
# that can be updated (the owner field cannot be updated).
@unittest.skip("Can't test the tag.update endpoint, "
"since there are no fields that can be updated")
def test_update(self):
""" Test that a tag can be updated """
# Update the tag using the OpenPhoto class, passing in the tag object

5
tox.ini
View file

@ -1,7 +1,10 @@
[tox]
envlist = py27,py26
envlist = py26, py27, py33
[testenv]
commands = python -m unittest discover --catch
[testenv:py26]
commands = unit2 discover --catch
deps =
unittest2