[feat] add first working use of pydantic input validation

events/payload_point2.json

@@ -9,5 +9,6 @@
     "label": 0
   }],
   "zoom": 10,
-  "source_type": "Satellite"
+  "source_type": "Satellite",
+  "debug": true
 }

requirements_dev.txt

@@ -7,5 +7,6 @@ numpy
 onnxruntime
 opencv-python
 pillow
+pydantic>=2.0.3
 rasterio
 requests

src/__init__.py

@@ -2,7 +2,9 @@ from aws_lambda_powertools import Logger
 import os
 from pathlib import Path
 
+from src.utilities.constants import SERVICE_NAME
+
 
 PROJECT_ROOT_FOLDER = Path(globals().get("__file__", "./_")).absolute().parent.parent
 MODEL_FOLDER = Path(os.path.join(PROJECT_ROOT_FOLDER, "models"))
-app_logger = Logger()
+app_logger = Logger(service=SERVICE_NAME)

src/app.py

@@ -1,18 +1,58 @@
 import json
 import time
 from http import HTTPStatus
-from typing import Dict
+from typing import Dict, List
 
 from aws_lambda_powertools.event_handler import content_types
 from aws_lambda_powertools.utilities.typing import LambdaContext
+from aws_lambda_powertools.utilities.parser import BaseModel
 
 from src import app_logger
 from src.io.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates
 from src.prediction_api.predictors import samexporter_predict
-from src.utilities.constants import CUSTOM_RESPONSE_MESSAGES
+from src.utilities.constants import CUSTOM_RESPONSE_MESSAGES, DEFAULT_LOG_LEVEL
 from src.utilities.utilities import base64_decode
 
 
+list_float = List[float]
+llist_float = List[list_float]
+
+
+class LatLngDict(BaseModel):
+    lat: float
+    lng: float
+
+
+class RawBBox(BaseModel):
+    ne: LatLngDict
+    sw: LatLngDict
+
+
+class RawPrompt(BaseModel):
+    type: str
+    data: LatLngDict
+    label: int = 0
+
+
+class RawRequestInput(BaseModel):
+    bbox: RawBBox
+    prompt: RawPrompt
+    zoom: int | float
+    source_type: str = "Satellite"
+
+
+class ParsedPrompt(BaseModel):
+    type: str
+    data: llist_float
+    label: int = 0
+
+
+class ParsedRequestInput(BaseModel):
+    bbox: llist_float
+    prompt: ParsedPrompt
+    zoom: int | float
+
+
 def get_response(status: int, start_time: float, request_id: str, response_body: Dict = None) -> str:
     """
     Return a response for frontend clients.
@@ -42,7 +82,7 @@ def get_response(status: int, start_time: float, request_id: str, response_body:
     return json.dumps(response)
 
 
-def get_parsed_bbox_points(request_input: Dict) -> Dict:
+def get_parsed_bbox_points(request_input: RawRequestInput) -> Dict:
     app_logger.info(f"try to parsing input request {request_input}...")
     bbox = request_input["bbox"]
     app_logger.debug(f"request bbox: {type(bbox)}, value:{bbox}.")
@@ -67,7 +107,7 @@ def get_parsed_bbox_points(request_input: Dict) -> Dict:
             raise ValueError("valid prompt type is only 'point'")
 
     app_logger.debug(f"bbox => {bbox}.")
-    app_logger.debug(f'## request_input-prompt updated => {request_input["prompt"]}.')
+    app_logger.debug(f'request_input-prompt updated => {request_input["prompt"]}.')
 
     app_logger.info(f"unpacking elaborated {request_input}...")
     return {
@@ -85,23 +125,7 @@ def lambda_handler(event: dict, context: LambdaContext):
         app_logger.info(f"event version: {event['version']}.")
 
     try:
-        app_logger.info(f"event:{json.dumps(event)}...")
-        app_logger.info(f"context:{context}...")
-
-        try:
-            body = event["body"]
-        except Exception as e_constants1:
-            app_logger.error(f"e_constants1:{e_constants1}.")
-            body = event
-
-        app_logger.debug(f"body, #1: {type(body)}, {body}...")
-
-        if isinstance(body, str):
-            body_decoded_str = base64_decode(body)
-            app_logger.debug(f"body_decoded_str: {type(body_decoded_str)}, {body_decoded_str}...")
-            body = json.loads(body_decoded_str)
-
-        app_logger.info(f"body, #2: {type(body)}, {body}...")
+        body = get_parsed_request_body(context, event)
 
         try:
             prompt_latlng = body["prompt"]
@@ -120,3 +144,28 @@ def lambda_handler(event: dict, context: LambdaContext):
 
     app_logger.info(f"response_dumped:{response}...")
     return response
+
+
+def get_parsed_request_body(context, event):
+    app_logger.info(f"event:{json.dumps(event)}...")
+    app_logger.info(f"context:{context}...")
+    try:
+        body = event["body"]
+    except Exception as e_constants1:
+        app_logger.error(f"e_constants1:{e_constants1}.")
+        body = event
+    app_logger.debug(f"body, #1: {type(body)}, {body}...")
+    if isinstance(body, str):
+        body_decoded_str = base64_decode(body)
+        app_logger.debug(f"body_decoded_str: {type(body_decoded_str)}, {body_decoded_str}...")
+        body = json.loads(body_decoded_str)
+    app_logger.info(f"body, #2: {type(body)}, {body}...")
+    try:
+        log_level = 'DEBUG' if body['debug'] else DEFAULT_LOG_LEVEL
+        app_logger.warning(f"set logger level to DEBUG")
+        app_logger.setLevel(log_level)
+    except KeyError:
+        app_logger.warning(f"can't set log level, reset it...")
+        app_logger.setLevel(DEFAULT_LOG_LEVEL)
+    app_logger.warning(f"logger level is {app_logger.log_level}.")
+    return body

src/utilities/constants.py

@@ -16,3 +16,5 @@ WKT_3857 = 'PROJCS["WGS 84 / Pseudo-Mercator",GEOGCS["WGS 84",DATUM["WGS_1984",S
 WKT_3857 += 'AUTHORITY["EPSG","8901"]],UNIT["degree",0.0174532925199433,AUTHORITY["EPSG","9122"]],AUTHORITY["EPSG","4326"]],PROJECTION["Mercator_1SP"],PARAMETER["central_meridian",0],'
 WKT_3857 += 'PARAMETER["scale_factor",1],PARAMETER["false_easting",0],PARAMETER["false_northing",0],UNIT["metre",1,AUTHORITY["EPSG","9001"]],AXIS["X",EAST],AXIS["Y",NORTH],EXTENSION["PROJ4",'
 WKT_3857 += '"+proj=merc +a=6378137 +b=6378137 +lat_ts=0.0 +lon_0=0.0 +x_0=0.0 +y_0=0 +k=1.0 +units=m +nadgrids=@null +wktext +no_defs"],AUTHORITY["EPSG","3857"]]'
+SERVICE_NAME = "sam-gis"
+DEFAULT_LOG_LEVEL = 'INFO'

src/utilities/type_hints.py

@@ -1,3 +1,13 @@
 """custom type hints"""
+from typing import TypedDict
+
+
 ts_dict_str2 = dict[str, str]
 ts_dict_str3 = dict[str, str, any]
+ts_ddict1 = dict[str, dict[str, any], dict, dict, any]
+# ts_list_float = list[float, float]
+
+
+class PixelCoordinate(TypedDict):
+    x: int
+    y: int