Python과 Terraform을 사용하여 이중 모드 AWS Lambda 구축

단계별 예시

Page content

여기에서 SQS 메시지 처리기의 Python Lambda 예제 + API 키 보호 기능이 있는 REST API와 함께 Terraform 스크립트를 사용하여 서버리스 실행을 위해 배포할 수 있습니다.

AWS Lambda는 거의 모든 이벤트에 반응할 수 있는 가벼운 서버리스 함수를 작성할 수 있게 해줍니다 — SQS 메시지부터 HTTP 요청까지. 이 가이드에서는 단일 Python Lambda를 사용하여 두 가지 모드로 작동하는 것을 구축할 것입니다:

  1. SQS 모드: SQS 메시지로 트리거될 때, 예를 들어 { "par": 10 }, 다른 대기열에 { "res": 11 }를 게시합니다.
  2. HTTP 모드: GET /lam?par=10에서 API Gateway를 통해 호출될 때, 클라이언트에게 { "res": 11 }을 반환합니다.

HTTP 엔드포인트는 간단한 하드코딩된 API 키 "testkey"를 사용하여 보호할 것입니다. 전체 설정은 Terraform을 사용하여 배포됩니다.

아키텍처 개요

우리가 구축하는 것을 시각화해 보겠습니다:

python lambda on aws sqs with api diagram

동일한 Lambda는 다음과 같이 반응합니다:

  • SQS 이벤트, 이벤트 소스 매핑을 통해,
  • API Gateway 요청, RESTful HTTP 통합을 통해.

단계 1: Python에서 Lambda 생성

SQS 이벤트와 HTTP API 호출 사이를 구분할 수 있는 매우 간단한 핸들러를 Python에서 생성해 보겠습니다.

파일: lambda_function.py

import json
import os
import boto3

sqs = boto3.client("sqs")

OUTPUT_QUEUE_URL = os.environ.get("OUTPUT_QUEUE_URL")
API_KEY = os.environ.get("API_KEY", "testkey")  # 하드코딩된 기본값

def lambda_handler(event, context):
    # 이벤트 유형 감지
    if "Records" in event:  # SQS 이벤트
        return handle_sqs(event["Records"])
    else:                   # HTTP 이벤트
        return handle_http(event)

def handle_sqs(records):
    for record in records:
        body = json.loads(record["body"])
        par = int(body["par"])
        res = par + 1
        message = json.dumps({"res": res})
        sqs.send_message(QueueUrl=OUTPUT_QUEUE_URL, MessageBody=message)
    return {"status": "processed", "count": len(records)}

def handle_http(event):
    headers = {k.lower(): v for k, v in (event.get("headers") or {}).items()}
    if headers.get("x-api-key") != API_KEY:
        return {
            "statusCode": 403,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps({"error": "Forbidden"})
        }

    params = event.get("queryStringParameters") or {}
    if "par" not in params:
        return {
            "statusCode": 400,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps({"error": "Missing par"})
        }

    par = int(params["par"])
    return {
        "statusCode": 200,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"res": par + 1})
    }

이 람다 함수에서 우리가 가지고 있는 것은 다음과 같습니다:

  • SQS 메시지는 JSON으로 파싱됩니다.
  • API Gateway에 의해 트리거될 때, 함수는 API 키와 쿼리 매개변수를 검증합니다.
  • 출력 대기열 URL과 API 키는 환경 변수를 통해 전달됩니다.

단계 2: Terraform으로 배포

Terraform은 AWS 인프라 — Lambda, SQS 대기열, API Gateway 및 권한 —을 한 번에 선언적으로 설정할 수 있게 해줍니다.

프로젝트 구조:

project/
├── lambda/
│   └── lambda_function.py
└── infra/
    └── main.tf

Terraform 구성 (infra/main.tf)

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

provider "aws" {
  region = "us-east-1"
}

locals {
  project = "lambda-sqs-api"
}

# Lambda 패키징
data "archive_file" "lambda_zip" {
  type        = "zip"
  source_dir  = "../lambda"
  output_path = "lambda.zip"
}

# SQS 대기열
resource "aws_sqs_queue" "input" {
  name = "${local.project}-input"
}

resource "aws_sqs_queue" "output" {
  name = "${local.project}-output"
}

# Lambda를 위한 IAM 역할
data "aws_iam_policy_document" "assume_role" {
  statement {
    actions = ["sts:AssumeRole"]
    principals {
      type        = "Service"
      identifiers = ["lambda.amazonaws.com"]
    }
  }
}

resource "aws_iam_role" "lambda_role" {
  name               = "${local.project}-role"
  assume_role_policy = data.aws_iam_policy_document.assume_role.json
}

resource "aws_iam_policy" "lambda_policy" {
  name = "${local.project}-policy"
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "sqs:SendMessage",
          "sqs:ReceiveMessage",
          "sqs:DeleteMessage",
          "sqs:GetQueueAttributes"
        ]
        Resource = "*"
      },
      {
        Effect = "Allow"
        Action = [
          "logs:CreateLogGroup",
          "logs:CreateLogStream",
          "logs:PutLogEvents"
        ]
        Resource = "*"
      }
    ]
  })
}

resource "aws_iam_role_policy_attachment" "lambda_policy_attach" {
  role       = aws_iam_role.lambda_role.name
  policy_arn = aws_iam_policy.lambda_policy.arn
}

# Lambda 함수
resource "aws_lambda_function" "func" {
  filename         = data.archive_file.lambda_zip.output_path
  function_name    = local.project
  role             = aws_iam_role.lambda_role.arn
  handler          = "lambda_function.lambda_handler"
  runtime          = "python3.12"
  environment {
    variables = {
      OUTPUT_QUEUE_URL = aws_sqs_queue.output.id
      API_KEY          = "testkey"
    }
  }
}

# 이벤트 소스 매핑 (SQS → Lambda)
resource "aws_lambda_event_source_mapping" "sqs_trigger" {
  event_source_arn  = aws_sqs_queue.input.arn
  function_name     = aws_lambda_function.func.arn
  batch_size        = 1
  enabled           = true
}

# API Gateway
resource "aws_api_gateway_rest_api" "api" {
  name = "${local.project}-api"
}

resource "aws_api_gateway_resource" "lam" {
  rest_api_id = aws_api_gateway_rest_api.api.id
  parent_id   = aws_api_gateway_rest_api.api.root_resource_id
  path_part   = "lam"
}

resource "aws_api_gateway_method" "get_lam" {
  rest_api_id   = aws_api_gateway_rest_api.api.id
  resource_id   = aws_api_gateway_resource.lam.id
  http_method   = "GET"
  authorization = "NONE"
  api_key_required = true
}

resource "aws_api_gateway_integration" "lambda_integration" {
  rest_api_id = aws_api_gateway_rest_api.api.id
  resource_id = aws_api_gateway_resource.lam.id
  http_method = aws_api_gateway_method.get_lam.http_method

  integration_http_method = "POST"
  type                    = "AWS_PROXY"
  uri                     = aws_lambda_function.func.invoke_arn
}

resource "aws_lambda_permission" "api_gateway" {
  statement_id  = "AllowAPIGatewayInvoke"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.func.function_name
  principal     = "apigateway.amazonaws.com"
  source_arn    = "${aws_api_gateway_rest_api.api.execution_arn}/*/*"
}

# API 키 및 사용 계획
resource "aws_api_gateway_api_key" "key" {
  name    = "testkey"
  value   = "testkey"
  enabled = true
}

resource "aws_api_gateway_usage_plan" "plan" {
  name = "basic"
  api_stages {
    api_id = aws_api_gateway_rest_api.api.id
    stage  = aws_api_gateway_deployment.deploy.stage_name
  }
}

resource "aws_api_gateway_usage_plan_key" "plan_key" {
  key_id        = aws_api_gateway_api_key.key.id
  key_type      = "API_KEY"
  usage_plan_id = aws_api_gateway_usage_plan.plan.id
}

resource "aws_api_gateway_deployment" "deploy" {
  depends_on = [aws_api_gateway_integration.lambda_integration]
  rest_api_id = aws_api_gateway_rest_api.api.id
  stage_name  = "v1"
}

output "api_url" {
  value = "${aws_api_gateway_deployment.deploy.invoke_url}/lam"
}

단계 3: 배포 및 테스트

  1. Terraform 초기화:
cd infra
terraform init
  1. 구성 적용:
terraform apply
  1. API Gateway 엔드포인트 테스트:
curl -H "x-api-key: testkey" "<API_URL>?par=10"
# 예상 결과: {"res": 11}
  1. SQS 테스트:

입력 대기열에 메시지 전송:

aws sqs send-message --queue-url <input-queue-url> --message-body '{"par": 5}'

그런 다음 출력 대기열 확인:

aws sqs receive-message --queue-url <output-queue-url>
# 예상 결과: {"res": 6}

단계 4: 정리

모든 리소스를 제거하려면:

terraform destroy

요약

[SQS 입력 대기열] ─▶ [Lambda 함수] ─▶ [SQS 출력 대기열]
                           ▲
                           │
                 [API Gateway /lam?par=N]
                           │
                     API 키로 보호됨

당신은 이제 다중 트리거 Lambda를 구축했습니다:

  • SQS 대기열에서 메시지를 소비하고 게시합니다.
  • API Gateway를 통해 HTTP 요청에 응답합니다.
  • 간단한 헤더 검사를 통해 API 키를 강제합니다.
  • Terraform을 통해 완전히 관리되는 서버리스 인프라를 위해 배포됩니다.

좋아요!

이 패턴은 가벼운 메시지 변환기, 하이브리드 마이크로서비스, 또는 비동기 및 동기 AWS 시스템을 연결하는 데 모두 몇 줄의 Python과 Terraform만으로 사용할 수 있습니다.

더 고급의 Lambda 예제를 보고 싶다면, 이 게시물을 확인해 보세요: AWS SAM + AWS SQS + Python PowerTools를 사용한 Lambda 코딩

유용한 링크