Library for caching requests to Nominatim in AWS DynamoDB

cachedgeo.py

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()