Skip to content

REST API

create_item()

Create a new item from the request body.

Returns:

Type Description
Response

201 with the created item, 422 on validation error, or 500 on error.

Source code in templates/api/handler.py
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@app.post("/items")
def create_item() -> Response:
    """Create a new item from the request body.

    Returns:
        201 with the created item, 422 on validation error, or 500 on error.
    """
    try:
        item = Item.model_validate_json(app.current_event.body)
    except ValidationError as exc:
        return JsonResponse({"errors": exc.errors()}, status_code=422)

    try:
        repository.put_item(item.model_dump())
    except Exception as exc:
        logger.error("DynamoDB put_item failed", exc_info=exc, extra={"itemId": item.id})
        return JsonResponse({"message": "Internal server error"}, status_code=500)

    return JsonResponse(item.dump_json(), status_code=201)

get_item(id)

Retrieve an item by ID.

Parameters:

Name Type Description Default
id str

The unique identifier of the item.

required

Returns:

Type Description
Response

200 with the item, 404 if not found, or 500 on error.

Source code in templates/api/handler.py
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@app.get("/items/<id>")
def get_item(id: str) -> Response:
    """Retrieve an item by ID.

    Args:
        id: The unique identifier of the item.

    Returns:
        200 with the item, 404 if not found, or 500 on error.
    """
    try:
        if (item := repository.get_item(id)) is None:
            return JsonResponse({"message": "Not found"}, status_code=404)
        item = Item.model_validate(item)  # Validate model after retrieval to ensure data integrity
    except Exception as exc:
        message = "Item validation failed" if isinstance(exc, ValidationError) else "Error retrieving item"
        logger.error(message, exc_info=exc, extra={"itemId": id})
        return JsonResponse({"message": "Internal server error"}, status_code=500)

    return JsonResponse(item.dump_json())

main(event, context)

Lambda entry point for the API Gateway handler.

Parameters:

Name Type Description Default
event dict

The API Gateway proxy event.

required
context LambdaContext

The Lambda execution context.

required

Returns:

Type Description
dict

The API Gateway proxy response.

Source code in templates/api/handler.py
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics
def main(event: dict, context: LambdaContext) -> dict:
    """Lambda entry point for the API Gateway handler.

    Args:
        event: The API Gateway proxy event.
        context: The Lambda execution context.

    Returns:
        The API Gateway proxy response.
    """
    return app.resolve(event, context)

JsonResponse

Bases: Response

An HTTP response with JSON body and security headers.

Source code in templates/api/response.py
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class JsonResponse(Response):
    """An HTTP response with JSON body and security headers."""

    def __init__(self, body: dict | str, status_code: int = 200) -> None:
        """Initialize the JSON response.

        Args:
            body: The response body as a dictionary or JSON string.
            status_code: The HTTP status code.
        """
        super().__init__(
            status_code=status_code,
            body=body if isinstance(body, str) else dumps(body),
            content_type="application/json",
            headers=SECURITY_HEADERS,
        )

__init__(body, status_code=200)

Initialize the JSON response.

Parameters:

Name Type Description Default
body dict | str

The response body as a dictionary or JSON string.

required
status_code int

The HTTP status code.

200
Source code in templates/api/response.py
15
16
17
18
19
20
21
22
23
24
25
26
27
def __init__(self, body: dict | str, status_code: int = 200) -> None:
    """Initialize the JSON response.

    Args:
        body: The response body as a dictionary or JSON string.
        status_code: The HTTP status code.
    """
    super().__init__(
        status_code=status_code,
        body=body if isinstance(body, str) else dumps(body),
        content_type="application/json",
        headers=SECURITY_HEADERS,
    )