Created
April 7, 2015 16:11
-
-
Save bbengfort/6d0c1004616794b0ed4f to your computer and use it in GitHub Desktop.
Lookup books on Amazon's API.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from django.conf import settings | |
from datetime import datetime | |
from urllib import urlencode | |
from httplib import HTTPConnection | |
from xml.dom.minidom import parse | |
import hmac, hashlib, base64 | |
ACCESS_ID = getattr(settings, 'AWS')['ID'] | |
ACCESS_SECRET = getattr(settings, 'AWS')['SECRET'] | |
class AWSException(Exception): | |
""" Abstract this to deal with AWS error callbacks from the API """ | |
pass | |
class ResponseNode(object): | |
def __init__(self, node): | |
self.node = node | |
def __repr__(self): | |
return "<%s %s>" % (self.__class__.__name__, self.node.nodeName) | |
def __str__(self): | |
if len(self.node.childNodes) == 1 and self.node.firstChild.nodeType == self.node.TEXT_NODE: | |
return self.node.firstChild.data | |
else: | |
return self.node.nodeName | |
def __getattr__(self, name): | |
nodes = [ResponseNode(n) for n in self.node.childNodes if n.nodeName == name] | |
if len(nodes) == 0: | |
return None | |
elif len(nodes) == 1: | |
if str(nodes[0]) == name: | |
return nodes[0] | |
else: | |
return str(nodes[0]) | |
else: | |
return nodes | |
def __len__(self): | |
return len(self.node.childNodes) | |
def __index__(self, i): | |
return ResponseNode(self.node.childNodes[i]) | |
def __iter__(self): | |
for node in self.node.childNodes: | |
yield ResponseNode(node) | |
def __contains__(self, name): | |
if getattr(self, name) is not None: | |
return True | |
return False | |
def __eq__(self, other): | |
if str(self) == str(other): | |
return True | |
return False | |
def children(self): | |
return [n.nodeName for n in self.node.childNodes] | |
def toxml(self): | |
return self.node.toxml( ) | |
def toprettyxml(self): | |
return self.node.toprettyxml( ) | |
class ResponseParser(ResponseNode): | |
""" | |
A factory class that returns Django objects from an AWS response XML. | |
""" | |
def __init__(self, dom): | |
super(ResponseParser, self).__init__(dom.documentElement) | |
if str(self.Items.Request.IsValid) == "False": | |
raise AWSException("Request Invalid") | |
elif "Errors" in self.Items.Request: | |
raise AWSException(self.Items.Request.Errors.Error.Message) | |
def books(self): | |
""" Constructs a dictionary of book attributes and returns them, temporary method for now """ | |
def findIsbn(obj): | |
tags = ['EAN', 'EISBN', 'ISBN', 'ASIN'] | |
for t in tags: | |
if getattr(obj, t, None) is not None: | |
return getattr(obj, t) | |
raise AWSException('ISBN not found!') | |
def findAuthors(obj): | |
if 'Author' in obj: | |
if isinstance(getattr(obj, 'Author'), (str, unicode)): | |
return getattr(obj, 'Author') | |
else: | |
return ', '.join([str(a) for a in getattr(obj, 'Author')]) | |
raise AWSException('Author(s) not found!') | |
if self.Items is not None: | |
for item in self.Items: | |
if item in ['Request', ]: continue | |
""" For Simplicity, I'm just grabbing the vars I hope are there... we'll do it more generically later. """ | |
yield { | |
'isbn': findIsbn(item.ItemAttributes), | |
'title': item.ItemAttributes.Title, | |
'pubdate': item.ItemAttributes.PublicationDate, | |
'publisher': item.ItemAttributes.Publisher, | |
'pages': item.ItemAttributes.NumberOfPages, | |
'authors': findAuthors(item.ItemAttributes), | |
'coverURL': item.LargeImage.URL, | |
'binding': item.ItemAttributes.Binding, | |
'amazonUrl': item.DetailPageURL, | |
} | |
class AWS(object): | |
VERB = 'GET' | |
HOST = 'ecs.amazonaws.com' | |
URI = '/onca/xml' | |
def __init__(self, **kwargs): | |
self.Service = 'AWSECommerceService' | |
self.Version = '2010-11-01' | |
self.AWSAccessKeyId = ACCESS_ID | |
self.Operation = None | |
self.ResponseGroup = ','.join(['Images','ItemAttributes',]) | |
# Will add any additional parameters to query or override defaults: | |
for (k,v) in kwargs.iteritems( ): | |
setattr(self, k, v) | |
@property | |
def Timestamp(self): | |
ts = datetime.utcnow( ) | |
return ts.strftime('%Y-%m-%dT%H:%M:%SZ') | |
def _signParams(self, params): | |
keys = params.keys( ) | |
keys.sort( ) # Sort in byte-order as required by the ECS API | |
sortedQuery = urlencode([(k, params[k]) for k in keys]) | |
signature = '\n'.join([self.VERB, self.HOST, self.URI, sortedQuery]) | |
dig = hmac.new(ACCESS_SECRET, msg=signature, digestmod=hashlib.sha256).digest( ) | |
params['Signature'] = base64.b64encode(dig).decode( ) | |
return params | |
def _requiredParams(self): | |
params = { } | |
# Add required parameters as stored in object: | |
for (k,v) in self.__dict__.iteritems( ): | |
params[k] = v | |
# Add dynamic required parameters: | |
params['Timestamp'] = self.Timestamp | |
return params | |
def _get(self, params): | |
# Create parameters string: | |
params.update(self._requiredParams( )) # Update with required params | |
params = self._signParams(params) # Sign the parameters | |
params = urlencode(params) # Urlencode the parameters | |
# Create the HTTP Connection | |
conn = HTTPConnection(self.HOST) | |
conn.request(self.VERB, '?'.join([self.URI, params])) | |
response = conn.getresponse( ) | |
data = parse(response) | |
conn.close( ) | |
return data | |
def itemLookup(self, isbn): | |
self.Operation = 'ItemLookup' | |
lookupParams = {'IdType':'ISBN', 'ItemId':isbn, 'SearchIndex':'Books'} | |
try: | |
return ResponseParser(self._get(lookupParams)) | |
except Exception as e: | |
raise AWSException(str(e)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Hey, I'm quite new to python and django as a whole. Can you please explain how to use this? I'm interested in getting book info from ISBN.