WebAPI Implementation Guide

Documentation describing practices for implementing WebAPI in this catalog. The key points are the following two:

• Main WebAPI functionality uses the official AWS library Powertools for AWS Lambda (Python)
• Adopts Lambdalith configuration as internal architecture
  • Superior in performance and operational convenience compared to separating Lambda resources for each API
  • Can follow the development configuration of traditional WebAPI frameworks

Implementation of API for Registering User Information

Implement an API that accepts user information input and registers new user data.

API Specification

Create a summary of specifications as follows.

• Normal case

  [Request sample]
  POST /users
  {
    "name": "Shigeo",
    "country": "Japan",
    "age": 30
  }
  
  [Response sample 1]
  200 OK
  {
    "name": "Shigeo",
    "country": "Japan",
    "age": 30
  }

• Error case (name not specified)

  [Request sample]
  POST /users
  {
    # "name": "Shigeo",
    "country": "Japan",
    "age": 30
  }
  
  [Response sample]
  400 Bad Request
  {
    "message": "Missing name. name is required query parameter."
  }

General Processing Flow

As a recommended configuration for Lambdalith, it consists of the following four layers, called in order from top to bottom.

Layer Name	Summary
API Routing	Calls mapped Actions
Action	Performs input validation and Service calls
Service	Main logic and business logic for API
Dao	Database access processing and query execution

Additionally, configure directories as needed, such as aws/ for separated AWS SDK implementations, lib/ for separated external libraries, and models/ for defining data modeling and schemas.

API Routing Implementation

When a request is received, an event object containing request information is passed to the Lambda function entry point (src/sample_api/index.ts). Use decorators provided by Powertools to distribute to appropriate Actions as follows.

Implementation Example

• src/sample_api/index.ts

from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
from aws_lambda_powertools.event_handler.exceptions import BadRequestError

# Instantiate app
app = APIGatewayHttpResolver()

# Lambda function entry point
def handler(event, context):
    return app.resolve(event, context)

# Map POST /users to this function
@app.post("/users")
def post_users():
    event = app.current_event
    return CreateUsersAction().handle(http_api_event=event)

# When BadRequestError is thrown during processing, this function is called
@app.exception_handler(BadRequestError)
def handle_400_bad_request_error(exception):
    return Response(status_code=400, content_type="application/json", body=json.dumps({"message": exception.msg}))

# When an unintended Exception is thrown during processing, this function is called
@app.exception_handler(Exception)
def handle_500_internal_server_error(e: Exception):
  # This sample code is included in the catalog AMI.

Action Implementation

"Action" is responsible for the following roles:

• Acts as the "entry" and "exit" points for API processing
• Performs input validation, Service execution, response data type organization, etc.
• Sometimes called the Controller layer

Implementation Example

• src/sample_api/actions/create_users_action.py

import json
from aws_lambda_powertools.event_handler.exceptions import BadRequestError
from sample_api.services.users_service import UsersService

class CreateUsersAction:

    def handle(self, http_api_event):
        # Extract request body
        req_body = json.loads(http_api_event.body)

        # Extract request body items
        name = req_body.get('name')
        country = req_body.get('country')
        age = req_body.get('age')

        # Perform input validation
        self.validate_required_params(name)
        # Additional checks needed: whether age is numeric type, whether body is None

        # Service instantiation and execution
        service = UsersService()
        service.create_users(name, country, age)

        # Build and return response information
        return {
            'statusCode': 200,
            'body': {
                'message': "Successfully created user data."
            }
        }

    def validate_required_params(self, name):
        if name is None:
            raise BadRequestError('Missing name. name is required query parameter.')

    def validate_body_is_not_empty(self, http_api_event):
      # This sample code is included in the catalog AMI.

    def validate_numeric_params(self, age):
      # This sample code is included in the catalog AMI.

    def is_numeric(self, val):
      # This sample code is included in the catalog AMI.

Service Implementation

"Service" is the part responsible for the main logic and business logic of the API.

Implementation Example

• src/sample_api/services/users_service.py

from aws_lambda_powertools import Logger
from sample_api.daos.users_dao import UsersDao

users_dao = UsersDao()

class UsersService:

    def create_users(self, name, country, age):
        # Implement processing for user creation
        # For example, if there are unique ID numbering rules, implement numbering processing
        # Once ready for registration, perform user information write processing to database via DAO
        new_users_record = users_dao.create_users(name, country, age)

        return new_users_record

DAO Implementation

"DAO" is responsible for processing that accesses the database and executes queries.

Implementation Example

• src/sample_api/actions/users_service.py

import os, psycopg
from psycopg.rows import dict_row


class UsersDao:
    def get_connection_info(self):
        # Load parameter information required for psycopg3 connection creation.
        # This sample code is included in the catalog AMI.

    def create_user(self, name, country, age):
        try:
            # Create database connection
            # In practice, please consider a connection pool mechanism.
            conn_info = self.get_connection_info()
            conn = psycopg.connect(**conn_info)
            with conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO users (name, country, age) VALUES (%s, %s, %s)",
                    (name, country, age)
                )
                conn.commit()
            conn.close()

Operation Verification

Please refer to "Local Debugging and Unit Testing".