handler.py

import json
import os
from decimal import Decimal

import boto3

dynamodb = boto3.resource("dynamodb")
s3 = boto3.client("s3")

TEST_RUNS_TABLE = os.environ["TEST_RUNS_TABLE"]
RESULTS_BUCKET = os.environ["RESULTS_BUCKET"]

table = dynamodb.Table(TEST_RUNS_TABLE)


def _json_default(value):
    if isinstance(value, Decimal):
        return int(value) if value % 1 == 0 else float(value)
    raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable")


def _response(status_code, body):
    return {
        "statusCode": status_code,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps(body, default=_json_default),
    }


def _list_tests(event):
    result = table.scan()
    return _response(200, {"tests": result.get("Items", [])})


def _result_keys(test_id):
    prefix = f"results/{test_id}/"
    paginator = s3.get_paginator("list_objects_v2")
    keys = []
    for page in paginator.paginate(Bucket=RESULTS_BUCKET, Prefix=prefix):
        keys.extend(obj["Key"] for obj in page.get("Contents", []))
    return keys


def _get_test(test_id):
    item = table.get_item(Key={"testId": test_id}).get("Item")
    if not item:
        return _response(404, {"message": "test not found"})
    return _response(200, {**item, "resultKeys": _result_keys(test_id)})


def lambda_handler(event, context):
    method = event["httpMethod"]
    resource = event["resource"]
    path_params = event.get("pathParameters") or {}
    test_id = path_params.get("testId")

    try:
        if resource == "/tests" and method == "GET":
            return _list_tests(event)
        if resource == "/tests/{testId}" and method == "GET":
            return _get_test(test_id)
        return _response(404, {"message": "not found"})
    except Exception as exc:  # noqa: BLE001 - surface as a 500 with a message
        return _response(500, {"message": str(exc)})