lambda_function.py

"""
This module defines an AWS Lambda function that acts as a proxy to the OS Places API,
appending ONS Geography data to the API response for each returned address result.
"""

import json
import os
import boto3
import requests

from catalyst_ons_geographies.postcodes import get_ons_from_postcodes

DB_PATH = '/tmp/ons_postcodes.duckdb'

# On init ensure the DuckDB database file exists in the Lambda temporary storage
s3 = boto3.client('s3')
s3_bucket = os.environ.get('ONSGEOGRAPHY_DB_BUCKET', 'geovation-catalyst')
s3_file = os.environ.get('ONSGEOGRAPHY_DB_FILE', 'data/ons_postcodes.duckdb')

if not os.path.exists(DB_PATH):
    s3.download_file(s3_bucket, s3_file, DB_PATH)


def lambda_handler(event, context):
    '''Lambda function to append ONS Geography data to OS API calls'''

    # Check if 'operation' key exists in pathParameters
    if 'pathParameters' in event and 'operation' in event['pathParameters'] and 'queryStringParameters' in event:
        operation = event['pathParameters']['operation']
    else:
        return {
            "statusCode": 400,
            "headers": {
                "Content-Type": "application/json"
            },
            "body": json.dumps({"error": "Missing path or query parameters"}),
            "isBase64Encoded": False
        }

    # Extract the query parameters
    query_params = event.get('queryStringParameters', {})

    # Construct the OS Places API URL
    api_url = f"https://api.os.uk/search/places/v1/{operation}?"

    # Add all the query parameters to the URL
    for key, value in query_params.items():
        api_url += f"&{key}={value}"

    try:
        # Call the OS Places API
        response = requests.get(api_url, timeout=10)

        # If the response is successful, append the ONS Geography data
        if response.status_code == 200:
            response_json = response.json()

            # Create an array of postcodes by trying first from the LPI object or the DPA object
            postcodes = []
            for result in response_json["results"]:
                postcode = result.get("LPI", {}).get(
                    "POSTAL_ADDRESS_CODE") or result.get("DPA", {}).get("POSTCODE")
                if postcode and postcode.replace(' ', '') not in postcodes:
                    postcodes.append(postcode.replace(' ', ''))

            ons_data_array = get_ons_from_postcodes(postcodes, DB_PATH)

            # For each result, append the ONS Geography data
            for result in response_json["results"]:
                postcode = result.get("DPA", {}).get("POSTCODE")
                ons_postcode_data = None
                for ons_data in ons_data_array:
                    if ons_data.get("postcode") == postcode.replace(' ', ''):
                        ons_postcode_data = ons_data
                        break
                result["ons_postcode_data"] = ons_postcode_data

            return {
                "statusCode": 200,
                "headers": {
                    "Content-Type": "application/json"
                },
                "body": json.dumps(response_json),
                "isBase64Encoded": False
            }

        else:
            return {
                "statusCode": response.status_code,
                "headers": {
                    "Content-Type": "text/plain"
                },
                "body": f"Error calling OS Places API: {response.text}",
                "isBase64Encoded": False
            }

    except Exception as e:
        return {
            "statusCode": 500,
            "headers": {
                "Content-Type": "text/plain"
            },
            "body": f"Error calling OS Places API: {repr(e)}",
            "isBase64Encoded": False
        }