Last active
March 24, 2024 17:49
-
-
Save tylerjw/81fb59b53da6c2de1f295f3f0e22d75c to your computer and use it in GitHub Desktop.
Library for caching requests to Nominatim in AWS DynamoDB
This file contains 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
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