File size: 1,953 Bytes
243f395 f2a79fa c6054f0 f2a79fa 243f395 c6054f0 f2a79fa c6054f0 f2a79fa c6054f0 f2a79fa c6054f0 f2a79fa |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
import json
import time
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.typing import LambdaContext
from pydantic import BaseModel, ValidationError
from src.utilities.type_hints import input_floatlist, input_floatlist2
logger = Logger()
class BBoxWithPointInput(BaseModel):
bbox: input_floatlist
points: input_floatlist2
def get_response(status: int, start_time: float, output: BaseModel = None) -> dict[str]:
"""
Return a response for frontend clients.
Args:
status: status response
start_time: request start time (float)
output: dict we embed into our response
Returns:
dict: response
"""
messages = {200: "ok", 422: "validation error", 500: "internal server error"}
body = {"duration_run": time.time() - start_time, "message": messages[status]}
if status == 200:
body = {"output": output.model_dump_json(), **body}
return {
"statusCode": status,
"headers": {'Content-type': 'application/json', 'Accept': 'application/json'},
"body": json.dumps(body)
}
def lambda_handler(event: dict, context: LambdaContext) -> dict[str]:
logger.info(f"start with aws_request_id:{context.aws_request_id}.")
start_time = time.time()
try:
logger.debug(f"event:{json.dumps(event)}...")
logger.debug(f"context:{context}...")
try:
bbox_points = BBoxWithPointInput(bbox=event["bbox"], points=event["points"])
logger.info(f"validation ok, bbox_points:{bbox_points}...")
response = get_response(200, start_time, bbox_points)
except ValidationError as ve:
logger.error(f"validation error:{ve}.")
response = get_response(422, start_time)
except Exception as e:
logger.error(f"exception:{e}.")
response = get_response(500, start_time)
logger.info(f"response:{response}...")
return response
|