"""
A simple wrapper around the 'experimental' Google Places API, documented
here: http://code.google.com/apis/maps/documentation/places/. This library
also makes use of the v3 Maps API for geocoding.
Prerequisites: A Google API key with Places activated against it. Please
check the Google API console, here: http://code.google.com/apis/console
NOTE: Please ensure that you read the Google terms of service (labelled 'Limits
and Requirements' on the documentation url) prior to using this library in a
production environment.
@author: sam@slimkrazy.com
"""
from __future__ import absolute_import
import cgi
from decimal import Decimal
try:
import json
except ImportError:
import simplejson as json
try:
import six
from six.moves import urllib
except ImportError:
pass
import warnings
from . import lang
from . import ranking
__all__ = ['GooglePlaces', 'GooglePlacesError', 'GooglePlacesAttributeError',
'geocode_location']
__version__ = '1.2.0'
__author__ = 'Samuel Adu'
__email__ = 'sam@slimkrazy.com'
class cached_property(object):
def __init__(self, func):
self.func = func
def __get__(self, instance, cls=None):
result = instance.__dict__[self.func.__name__] = self.func(instance)
return result
def _fetch_remote(service_url, params={}, use_http_post=False):
encoded_data = {}
for k, v in params.items():
if isinstance(v, six.string_types):
v = v.encode('utf-8')
encoded_data[k] = v
encoded_data = urllib.parse.urlencode(encoded_data)
if not use_http_post:
query_url = (service_url if service_url.endswith('?') else
'%s?' % service_url)
request_url = query_url + encoded_data
request = urllib.request.Request(request_url)
else:
request_url = service_url
request = urllib.request.Request(service_url, data=encoded_data)
return (request_url, urllib.request.urlopen(request))
def _fetch_remote_json(service_url, params={}, use_http_post=False):
"""Retrieves a JSON object from a URL."""
request_url, response = _fetch_remote(service_url, params, use_http_post)
if six.PY3:
str_response = response.read().decode('utf-8')
return (request_url, json.loads(str_response, parse_float=Decimal))
return (request_url, json.load(response, parse_float=Decimal))
def _fetch_remote_file(service_url, params={}, use_http_post=False):
"""Retrieves a file from a URL.
Returns a tuple (mimetype, filename, data)
"""
request_url, response = _fetch_remote(service_url, params, use_http_post)
dummy, params = cgi.parse_header(
response.headers.get('Content-Disposition', ''))
fn = params['filename']
return (response.headers.get('content-type'),
fn, response.read(), response.geturl())
def geocode_location(location, sensor=False):
"""Converts a human-readable location to lat-lng.
Returns a dict with lat and lng keys.
keyword arguments:
location -- A human-readable location, e.g 'London, England'
sensor -- Boolean flag denoting if the location came from a device using
its' location sensor (default False)
raises:
GooglePlacesError -- if the geocoder fails to find a location.
"""
url, geo_response = _fetch_remote_json(
GooglePlaces.GEOCODE_API_URL,
{'address': location, 'sensor': str(sensor).lower()})
_validate_response(url, geo_response)
if geo_response['status'] == GooglePlaces.RESPONSE_STATUS_ZERO_RESULTS:
error_detail = ('Lat/Lng for location \'%s\' can\'t be determined.' %
location)
raise GooglePlacesError(error_detail)
return geo_response['results'][0]['geometry']['location']
def _get_place_details(place_id, api_key, sensor=False,
language=lang.ENGLISH):
"""Gets a detailed place response.
keyword arguments:
place_id -- The unique identifier for the required place.
"""
url, detail_response = _fetch_remote_json(GooglePlaces.DETAIL_API_URL,
{'placeid': place_id,
'sensor': str(sensor).lower(),
'key': api_key,
'language': language})
_validate_response(url, detail_response)
return detail_response['result']
def _get_place_photo(photoreference, api_key, maxheight=None, maxwidth=None,
sensor=False):
"""Gets a place's photo by reference.
See detailed documentation at https://developers.google.com/places/documentation/photos
Arguments:
photoreference -- The unique Google reference for the required photo.
Keyword arguments:
maxheight -- The maximum desired photo height in pixels
maxwidth -- The maximum desired photo width in pixels
You must specify one of this keyword arguments. Acceptable value is an
integer between 1 and 1600.
"""
params = {'photoreference': photoreference,
'sensor': str(sensor).lower(),
'key': api_key}
if maxheight:
params['maxheight'] = maxheight
if maxwidth:
params['maxwidth'] = maxwidth
return _fetch_remote_file(GooglePlaces.PHOTO_API_URL, params)
def _validate_response(url, response):
"""Validates that the response from Google was successful."""
if response['status'] not in [GooglePlaces.RESPONSE_STATUS_OK,
GooglePlaces.RESPONSE_STATUS_ZERO_RESULTS]:
error_detail = ('Request to URL %s failed with response code: %s' %
(url, response['status']))
raise GooglePlacesError(error_detail)
class GooglePlacesError(Exception):
pass
class GooglePlacesAttributeError(AttributeError):
"""Exception thrown when a detailed property is unavailable.
A search query from the places API returns only a summary of the Place.
in order to get full details, a further API call must be made using
the place_id. This exception will be thrown when a property made
available by only the detailed API call is looked up against the summary
object.
An explicit call to get_details() must be made on the summary object in
order to convert a summary object to a detailed object.
"""
# I could spend forever muling between this design decision and creating
# a PlaceSummary object as well as a Place object. I'm leaning towards this
# method in order to keep the API as simple as possible.
pass
class GooglePlaces(object):
"""A wrapper around the Google Places Query API."""
BASE_URL = 'https://maps.googleapis.com/maps/api'
PLACE_URL = BASE_URL + '/place'
GEOCODE_API_URL = BASE_URL + '/geocode/json?'
RADAR_SEARCH_API_URL = PLACE_URL + '/radarsearch/json?'
NEARBY_SEARCH_API_URL = PLACE_URL + '/nearbysearch/json?'
TEXT_SEARCH_API_URL = PLACE_URL + '/textsearch/json?'
AUTOCOMPLETE_API_URL = PLACE_URL + '/autocomplete/json?'
DETAIL_API_URL = PLACE_URL + '/details/json?'
CHECKIN_API_URL = PLACE_URL + '/check-in/json?sensor=%s&key=%s'
ADD_API_URL = PLACE_URL + '/add/json?sensor=%s&key=%s'
DELETE_API_URL = PLACE_URL + '/delete/json?sensor=%s&key=%s'
PHOTO_API_URL = PLACE_URL + '/photo?'
MAXIMUM_SEARCH_RADIUS = 50000
RESPONSE_STATUS_OK = 'OK'
RESPONSE_STATUS_ZERO_RESULTS = 'ZERO_RESULTS'
def __init__(self, api_key):
self._api_key = api_key
self._sensor = False
self._request_params = None
def query(self, **kwargs):
with warnings.catch_warnings():
warnings.simplefilter('always')
warnings.warn('The query API is deprecated. Please use nearby_search.',
DeprecationWarning, stacklevel=2)
return self.nearby_search(**kwargs)
def nearby_search(self, language=lang.ENGLISH, keyword=None
