基本的python代碼

refs

app.py

def handler(event, context):
    print('event', event)
    return {
        'message': "hello lambda"
    }

Dockerfile

ref

FROM public.ecr.aws/lambda/python:3.9

COPY app.py ${LAMBDA_TASK_ROOT}

CMD [ "app.handler" ]

Docker Build/Push

需要先建立好ECR Repository

並確認AWS CLI擁有ECR權限

docker build -t lambda-python .

docker tag lambda-python:latest {ECR_REPOSITORY_URI}:latest

docker push {ECR_REPOSITORY_URI}:latest

本機測試

先將服務啟動

docker run -p 9000:8080 lambda-python

使用curl測試

curl -XPOST "http://localhost:9000/2015-03-31/functions/function/invocations" -d '{"foo": "bar"}'

服務Log

執行輸出

Lambda測試結果

使用Fastapi佈署至Lambda整合API Gateway

這段會使用Mangum python套件來處理來自API Gateway的請求

Mangum會將將API Gateway的請求都處理好可以直接丟給fastapi

refs

參考影片 - Deploy FastAPI on AWS Lambda ⚡ Serverless hosting!

main.py

建議一個簡易的fastapi程式碼main.py

並安裝使用Mangum套件

from fastapi import FastAPI
from mangum import Mangum

app = FastAPI()


@app.get("/")
def index():
    return {"Hello": "lambda", "function": "index"}


@app.get("/hello")
def hello():
    return {"Hello": "lambda", "function": "hello"}


@app.get("/test")
def hello():
    return {"Hello": "lambda", "function": "test"}

handler = Mangum(app, lifespan="off")

Dockerfile

FROM public.ecr.aws/lambda/python:3.9

COPY main.py ${LAMBDA_TASK_ROOT}
COPY requirements.txt  .

RUN  pip3 install -r requirements.txt --target "${LAMBDA_TASK_ROOT}"

CMD [ "main.handler" ]

在本機測試

docker image build完之後可以在本機先執行起來測試

因為要模擬AWS API Gateway的AWS Proxy請求

所以會有這麼長的json要帶

可依照測試情況調整method及路由(共有3處的路由要調整)

這邊的路由可以直接對到fastapi的api路由

curl curl -XPOST "http://localhost:9000/2015-03-31/functions/function/invocations" \
  -d '{
  "body": "eyJ0ZXN0IjoiYm9keSJ9",
  "resource": "/{proxy+}",
  "path": "/hello",
  "httpMethod": "GET",
  "isBase64Encoded": true,
  "queryStringParameters": {
    "foo": "bar"
  },
  "multiValueQueryStringParameters": {
    "foo": [
      "bar"
    ]
  },
  "pathParameters": {
    "proxy": "/hello"
  },
  "stageVariables": {
    "baz": "qux"
  },
  "headers": {
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
    "Accept-Encoding": "gzip, deflate, sdch",
    "Accept-Language": "en-US,en;q=0.8",
    "Cache-Control": "max-age=0",
    "CloudFront-Forwarded-Proto": "https",
    "CloudFront-Is-Desktop-Viewer": "true",
    "CloudFront-Is-Mobile-Viewer": "false",
    "CloudFront-Is-SmartTV-Viewer": "false",
    "CloudFront-Is-Tablet-Viewer": "false",
    "CloudFront-Viewer-Country": "US",
    "Host": "1234567890.execute-api.us-east-1.amazonaws.com",
    "Upgrade-Insecure-Requests": "1",
    "User-Agent": "Custom User Agent String",
    "Via": "1.1 08f323deadbeefa7af34d5feb414ce27.cloudfront.net (CloudFront)",
    "X-Amz-Cf-Id": "cDehVQoZnx43VYQb9j2-nvCh-9z396Uhbp027Y2JvkCPNLmGJHqlaA==",
    "X-Forwarded-For": "127.0.0.1, 127.0.0.2",
    "X-Forwarded-Port": "443",
    "X-Forwarded-Proto": "https"
  },
  "multiValueHeaders": {
    "Accept": [
      "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8"
    ],
    "Accept-Encoding": [
      "gzip, deflate, sdch"
    ],
    "Accept-Language": [
      "en-US,en;q=0.8"
    ],
    "Cache-Control": [
      "max-age=0"
    ],
    "CloudFront-Forwarded-Proto": [
      "https"
    ],
    "CloudFront-Is-Desktop-Viewer": [
      "true"
    ],
    "CloudFront-Is-Mobile-Viewer": [
      "false"
    ],
    "CloudFront-Is-SmartTV-Viewer": [
      "false"
    ],
    "CloudFront-Is-Tablet-Viewer": [
      "false"
    ],
    "CloudFront-Viewer-Country": [
      "US"
    ],
    "Host": [
      "0123456789.execute-api.us-east-1.amazonaws.com"
    ],
    "Upgrade-Insecure-Requests": [
      "1"
    ],
    "User-Agent": [
      "Custom User Agent String"
    ],
    "Via": [
      "1.1 08f323deadbeefa7af34d5feb414ce27.cloudfront.net (CloudFront)"
    ],
    "X-Amz-Cf-Id": [
      "cDehVQoZnx43VYQb9j2-nvCh-9z396Uhbp027Y2JvkCPNLmGJHqlaA=="
    ],
    "X-Forwarded-For": [
      "127.0.0.1, 127.0.0.2"
    ],
    "X-Forwarded-Port": [
      "443"
    ],
    "X-Forwarded-Proto": [
      "https"
    ]
  },
  "requestContext": {
    "accountId": "123456789012",
    "resourceId": "123456",
    "stage": "prod",
    "requestId": "c6af9ac6-7b61-11e6-9a41-93e8deadbeef",
    "requestTime": "09/Apr/2015:12:34:56 +0000",
    "requestTimeEpoch": 1428582896000,
    "identity": {
      "cognitoIdentityPoolId": null,
      "accountId": null,
      "cognitoIdentityId": null,
      "caller": null,
      "accessKey": null,
      "sourceIp": "127.0.0.1",
      "cognitoAuthenticationType": null,
      "cognitoAuthenticationProvider": null,
      "userArn": null,
      "userAgent": "Custom User Agent String",
      "user": null
    },
    "path": "/prod/hello",
    "resourcePath": "/{proxy+}",
    "httpMethod": "POST",
    "apiId": "1234567890",
    "protocol": "HTTP/1.1"
  }
}'

在Lambda上測試

在lambda的測試可以選API Gateway AWS Proxy範本

一樣可調整Method及路由來測試

設定API Gateway

API Gateway設定很簡單

不用依照fastapi實際的所有api一個一個建立在API Gateway上

API Gateway提供一個Lambda Proxy Resource的方式來整合

只要建立Resource的時候勾選代理資源

接著整合類型選擇Lambda Function並設定好對應的Function即可

儲存後API Gateway的路由就可以直接對應所有Lambda中fastapi的路由

API Root設定

前面的設定除了root api之外

所有的規則都可以直接導向Lambda(例如: /foo, /bar, /foo/bar, /a/b/c)

但root需要額外設定

如果需要root api可以被使用的話必須設定Root的Get Method

然後一樣要設定Lambda及使用Lambda Proxy

使用aws python client library的注意事項

若要存取其他AWS服務

通常會用boto3套件來開發

在lambda上有些要注意的事項

因為在Lambda上會自動賦予一個Role(在建立Lambda Function的時候設定的Role)

而在Lambda上使用boto3套件的時候

會自動使用這個Role取得臨時憑證(STS Token)來存取AWS服務

因此如果在其他環境如果有使用IAM Key來存取AWS資源的話

需要分兩種模式

使用Assume Role的方式透過STS Token存取
透過IAM Key存取

範例

以下為一段建立S3物件準備存取S3服務的範例

這邊可以看到透過settings(pydantic settings)來依照環境變數確認當下是否在Lambda中

如果是在Lambda中就不設定IAM Key直接使用Assume Role STS Token

如果不是在Lambda中就直接使用IAM Key

class S3Service:
    def __init__(self):
        if settings.IN_LAMBDA is True:
            self.s3 = boto3.client('s3')
        else:
            session = boto3.Session(
                aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
                aws_secret_access_key=settings.AWS_ACCESS_KEY_SECRET,
            )
            self.s3 = session.client('s3')

在Lambda中使用IAM Key

如果直接在Lambda中使用IAM Key存取

仍無法正常使用

會發現IAM Key仍會被Assume Role的STS Token蓋掉

Terraform

如果是使用上述Mangum套件的情況加上使用Docker類型的Lambda Function

通常就是會需要ECR+API Gateway+Lambda+CloudWatch這些資源

這邊提供一個完整的terraform應用程式範例

透過terraform建立以下資源、設定

ECR Repository
API Gateway
- API(Index API、Proxy API)
- Lambda Proxy
- Usage Plan
- API Key
- Stage
- Deploy
Lambda
CloudWatch
IAM
- Lambda assume role

執行terraform的AWS CLI IAM權限

IAM FullAccess
API Gateway FullAccess
Lambda FullAccess
ECR FullAccess
CloudWatch FullAccess(調整Log Group retention time)

main.tf

variable "aws_profile" {
  type = string
  default = "default"
}

variable "aws_region" {
  type = string
  default = "us-east-1"
}

variable "service_name" {
  type = string
}

variable "ecr_repo_name" {
  type = string
}

variable "api_gateway_stage_name" {
  type = string
  default = "dev"
}

variable "api_key" {
  type = string
  default = "abced12345abced12345"
}

data "aws_caller_identity" "current" {}

locals {
  aws_account_id = data.aws_caller_identity.current.account_id
  ecr_host = "${local.aws_account_id}.dkr.ecr.${var.aws_region}.amazonaws.com"
  lambda_image_uri = "${local.ecr_host}/${var.ecr_repo_name}:latest"
}

terraform {
  required_providers {
    aws = {
      source = "hashicorp/aws"
      version = "~> 4.16"
    }
  }

  required_version = ">= 1.2.0"
}

provider "aws" {
  region = "ap-southeast-1"
  profile = var.aws_profile
}

ECR

這邊會建立一個ECR Repository

並立即push一個最簡單的alpine image

原因是要建立docker lambda function一開始就必須指定一個image

因此為了能夠順利建立lambda function

一開始會先push一個image

resource "aws_ecr_repository" "ecr" {
  name = var.ecr_repo_name
  force_delete = true
}

resource "null_resource" "build_push_image" {
  depends_on = [aws_ecr_repository.ecr]
  provisioner "local-exec" {
    command = templatefile("${path.module}/build_push_image.sh", {
      IMAGE_TAG = local.lambda_image_uri
      ECR_HOST = local.ecr_host
      REGION = var.aws_region
    })
  }
}

build_push_image.sh如下

echo "FROM alpine" > Dockerfile
aws ecr get-login-password --region ${REGION} | docker login --username AWS --password-stdin ${ECR_HOST}
docker build -t ${IMAGE_TAG} .
docker push ${IMAGE_TAG}

Lambda

# 設定cloud watch log retention
resource "aws_cloudwatch_log_group" "log_group" {
  name              = "/aws/lambda/${aws_lambda_function.lambda-function.function_name}"
  retention_in_days = 3
}

# lambda function
resource "aws_lambda_function" "lambda-function" {
  depends_on = [null_resource.build_push_image]
  function_name = var.service_name
  role = aws_iam_role.lambda-assume-role.arn
  package_type = "Image"
  image_uri = local.lambda_image_uri
  environment {
    variables = {
      MODE: "lambda",
    }
  }
}

# lambda assume role設定
resource "aws_iam_role" "lambda-assume-role" {
  name = "lambda-${var.service_name}-iam"
  assume_role_policy = <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}
EOF
}

# 設定lambda使用的role有基本的cloud watch寫入權限(AWSLambdaBasicExecutionRole), 否則無法寫入log至cloudwatch
resource "aws_iam_role_policy_attachment" "lambda_policy_attachment" {
  role = aws_iam_role.lambda-assume-role.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
}

# 設定lambda使用的role有基本的S3完整權限
resource "aws_iam_role_policy_attachment" "lambda_role_s3_fullaccess_attachment" {
  policy_arn = "arn:aws:iam::aws:policy/AmazonS3FullAccess"
  role = aws_iam_role.lambda-assume-role.name
}

# ref: https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_permission
# 允許api gateway存取lambda
resource "aws_lambda_permission" "apigw_lambda" {
  statement_id  = "AllowExecutionFromAPIGateway"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.lambda-function.function_name
  principal     = "apigateway.amazonaws.com"

  # More: http://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-control-access-using-iam-policies-to-invoke-api.html
  source_arn = "arn:aws:execute-api:${var.aws_region}:${local.aws_account_id}:${aws_api_gateway_rest_api.service.id}/*/*"
}

API Gateway

# ref: https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/api_gateway_rest_api
resource "aws_api_gateway_rest_api" "service" {
  name = var.service_name
  endpoint_configuration {
    types = ["REGIONAL"]
  }
}

# index method
resource "aws_api_gateway_method" "index" {
  rest_api_id = aws_api_gateway_rest_api.service.id
  resource_id = aws_api_gateway_rest_api.service.root_resource_id
  http_method = "GET"
  api_key_required = true
  authorization = "NONE"
}

resource "aws_api_gateway_integration" "index" {
  rest_api_id = aws_api_gateway_rest_api.service.id
  resource_id = aws_api_gateway_rest_api.service.root_resource_id
  http_method = "GET"
  type        = "AWS_PROXY"
  uri         = aws_lambda_function.lambda-function.invoke_arn
  integration_http_method = "POST"
  connection_type         = "INTERNET"
  lifecycle {
    ignore_changes = [connection_type]
  }
}

# proxy
resource "aws_api_gateway_resource" "proxy" {
  rest_api_id = aws_api_gateway_rest_api.service.id
  parent_id   = aws_api_gateway_rest_api.service.root_resource_id
  path_part   = "{proxy+}"
}

resource "aws_api_gateway_method" "proxy" {
  rest_api_id   = aws_api_gateway_rest_api.service.id
  resource_id   = aws_api_gateway_resource.proxy.id
  http_method   = "ANY"
  authorization = "NONE"
  api_key_required = true
}

resource "aws_api_gateway_integration" "proxy" {
  rest_api_id = aws_api_gateway_rest_api.service.id
  resource_id = aws_api_gateway_resource.proxy.id
  http_method = aws_api_gateway_method.proxy.http_method
  type        = "AWS_PROXY"
  uri         = aws_lambda_function.lambda-function.invoke_arn
  integration_http_method = "POST"
  connection_type         = "INTERNET"
  lifecycle {
    ignore_changes = [connection_type]
  }
}

# deploy
resource "aws_api_gateway_deployment" "deploy" {
  rest_api_id = aws_api_gateway_rest_api.service.id

  triggers = {
    redeployment = sha1(jsonencode([
      aws_api_gateway_method.index.id,
      aws_api_gateway_integration.index.id,
      aws_api_gateway_resource.proxy.id,
      aws_api_gateway_method.proxy.id,
      aws_api_gateway_integration.proxy.id,
    ]))
  }

  lifecycle {
    create_before_destroy = true
  }
}

resource "aws_api_gateway_stage" "api_gateway_stage" {
  deployment_id = aws_api_gateway_deployment.deploy.id
  rest_api_id   = aws_api_gateway_rest_api.service.id
  stage_name    = var.api_gateway_stage_name
}

# api key/plan
resource "aws_api_gateway_usage_plan" "plan" {
  name         = "${var.service_name}-plan"
  product_code = "MYCODE"

  api_stages {
    api_id = aws_api_gateway_rest_api.service.id
    stage  = aws_api_gateway_stage.api_gateway_stage.stage_name
  }

  quota_settings {
    limit  = 1000
    offset = 1
    period = "MONTH"
  }

  throttle_settings {
    burst_limit = 300
    rate_limit  = 10
  }
}

resource "aws_api_gateway_api_key" "api_key" {
  depends_on = [aws_api_gateway_usage_plan.plan]
  name = "${var.service_name}-api-key"
  value = var.api_key
}

resource "aws_api_gateway_usage_plan_key" "plan_key_bind" {
  depends_on = [aws_api_gateway_api_key.api_key]
  key_id        = aws_api_gateway_api_key.api_key.id
  key_type      = "API_KEY"
  usage_plan_id = aws_api_gateway_usage_plan.plan.id
}

# enable error logs
resource "aws_api_gateway_method_settings" "enable_error_log" {
  rest_api_id = aws_api_gateway_rest_api.service.id
  stage_name  = aws_api_gateway_stage.api_gateway_stage.stage_name
  method_path = "*/*"

  settings {
    logging_level = "ERROR"
  }
}

output "api_gateway_url" {
  value = aws_api_gateway_stage.api_gateway_stage.invoke_url
}

在AWS Lambda中使用python docker