Skip to content

Instantly share code, notes, and snippets.

@tmaybe
Last active June 14, 2016 18:49
Show Gist options
  • Save tmaybe/9519801 to your computer and use it in GitHub Desktop.
Save tmaybe/9519801 to your computer and use it in GitHub Desktop.
sample code for caching API requests in GAE
application: apipemilucache
version: 0-1-3
runtime: python27
api_version: 1
threadsafe: true
handlers:
- url: /.*
script: cache.application
from google.appengine.api import memcache
from google.appengine.api import urlfetch
import logging
import re
import pickle
import webapp2
# how to use logging (also works: logging.error('blah'))
# https://developers.google.com/api-client-library/python/guide/logging
# store values in memcache, splitting values > 1MB if necessary
# adapted from http://stackoverflow.com/a/9143912/958481
def store_chunks(key, value, cachetime=86400, chunksize=950000):
serialized = pickle.dumps(value, 2)
values = {}
for i in xrange(0, len(serialized), chunksize):
values['%s.%s' % (key, i//chunksize)] = serialized[i : i+chunksize]
keys_not_set = memcache.set_multi(values, time=cachetime)
if len(keys_not_set):
logging.error('memcache.set_multi failed for keys ' + ','.join(keys_not_set))
# if some keys were set, delete their values
if len(values) > len(keys_not_set):
logging.error('memcache.set_multi stored partial data; deleting cache for keys ' + ','.join(values.keys()))
memcache.delete_multi(values.keys())
return False
return True
# retrieve values from memcache, reconstituting values > 1MB if necessary
# adapted from http://stackoverflow.com/a/9143912/958481
def retrieve_chunks(key):
chunklist = ['%s.%s' % (key, i) for i in xrange(32)]
result = memcache.get_multi(['%s.%s' % (key, i) for i in xrange(32)])
serialized = ''.join([v for k, v in sorted(result.items()) if v is not None])
if serialized == '':
return None
return pickle.loads(serialized)
class MainPage(webapp2.RequestHandler):
def get(self):
# access to the parts of the URL that loaded this script:
# https://developers.google.com/appengine/docs/python/tools/webapp/requestclass
# ignore favicon.ico requests
if self.request.path != '/favicon.ico':
base = "http://api.pemiluapi.org"
route = self.request.path
if self.request.query_string != '':
route = route + '?' + self.request.query_string
url = base + route
# create a key for memcache that excludes the apiKey
params_list = [qu for qu in re.split('&', self.request.query_string) if 'apiKey' not in qu]
params_list.sort()
memkey = self.request.path + '?' + '&'.join(params_list)
# serve cached content if it exists
cached_content = retrieve_chunks(memkey)
if cached_content is not None:
self.response.headers['Content-Type'] = 'application/json'
self.response.headers['Access-Control-Allow-Origin'] = '*'
self.response.write(cached_content)
# no cached content exists; get it from API Pemilu
else:
# set timeout deadline to maximum (60) from default (5)
# https://developers.google.com/appengine/docs/python/urlfetch/fetchfunction
result = urlfetch.fetch(url, deadline=60)
# the result object from urlfetch.fetch():
# https://developers.google.com/appengine/docs/python/urlfetch/responseobjects
# only save good responses in memcache
if result.status_code == 200:
# :TODO: re-try if failed?
store_chunks(memkey, result.content)
# log API errors before sending them on
else:
logging.error('API returned error ' + str(result.status_code) + ' for URL ' + url)
# return the response
self.response.set_status(result.status_code)
self.response.headers['Content-Type'] = result.headers['Content-Type']
self.response.headers['Access-Control-Allow-Origin'] = '*'
self.response.write(result.content)
application = webapp2.WSGIApplication([
('/.*', MainPage),
], debug=False)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment