Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
Library for caching requests to Nominatim in AWS DynamoDB
import boto3
from botocore.exceptions import ClientError
import decimal
from putItem import addr_to_query, query_to_addr
from geopy.geocoders import Nominatim
import json
import logging
logger = logging.getLogger(__name__)
# Helper class to convert a DynamoDB item to JSON.
class DecimalEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, decimal.Decimal):
if o % 1 > 0:
return float(o)
else:
return int(o)
return super(DecimalEncoder, self).default(o)
def get_query_from_db(query):
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('geosearch')
try:
response = table.get_item(
Key={
'query': query
}
)
except ClientError as e:
logger.debug(e.response['Error']['Message'])
else:
if 'Item' in response:
item = json.dumps(response['Item'], indent=4, cls=DecimalEncoder)
return item
else:
return None
def push_address_to_db(address, query):
geolocator = Nominatim()
location = geolocator.geocode(address, timeout=10)
record = location_to_record(query,location)
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('geosearch')
table.put_item(Item=record)
def addr_to_query(addr):
return addr.replace(', ','--').replace(' ','-')
def query_to_addr(query):
return query.replace('--',', ').replace('-',' ')
def float_to_dec(val):
return decimal.Decimal(str(val))
def location_to_record(query,location):
return {
'query': query,
'address': location.address,
'latitude': float_to_dec(location.latitude),
'longitude': float_to_dec(location.longitude),
}
def get_address(address):
query = addr_to_query(address)
db_entry = get_query_from_db(query)
if db_entry is not None:
return db_entry
push_address_to_db(address, query)
return get_query_from_db(query)
def get_query(query):
address = query_to_addr(query)
return get_address(address)
# run this once to create the table the first time
def create_table():
dynamodb = boto3.resource('dynamodb')
table = dynamodb.create_table(
TableName='geosearch',
KeySchema=[
{
'AttributeName': 'query',
'KeyType': 'HASH'
}
],
AttributeDefinitions=[
{
'AttributeName': 'query',
'AttributeType': 'S'
}
],
ProvisionedThroughput={
'ReadCapacityUnits': 1,
'WriteCapacityUnits': 1
}
)
def main():
address = '1600 Pennsylvania Avenue NW Washington, DC 20500'
print(get_address(address))
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment