AWS SAM + AWS SQS + Python PowerTools
구현 예시
Page content
여기에서 AWS SAM에서 람다 함수를 구현하기 위한 단계를 나열합니다:
- REST API 람다: SQS에 메시지 보내기
- SQS 람다 트리거: SQS에서 메시지 처리하기
AWS SAM이 무엇인지 그리고 어떻게 시작하는지에 대해 궁금하다면 이 문서를 확인하세요: AWS SAM과 파이썬을 사용한 계층형 람다
사전 설정
AWS SAM과 파이썬을 사용하여 REST-API를 구현합니다. 예를 들어, AWS SAM과 파이썬을 사용한 계층형 람다에서 설명한 것처럼 구현할 수 있습니다.
REST API 람다: SQS에 메시지 보내기
template.yaml에 큐를 설명합니다.
MySQSQueue:
Type: String
Default: "TheQueue"
MySQSQueueArn:
Type: String
Default: "arn:aws:sqs:ap-southeast-2:123somenumbers123:TheQueue"
AWS SQS로 메시지를 보내는 API 함수를 template.yaml에 추가합니다.
PostMessageFunction:
Type: AWS::Serverless::Function
Properties:
Handler: app.lambda_handler
CodeUri: test
MemorySize: 1024
Description: PostMessageFunction 함수
Runtime: python3.12
Architectures:
- x86_64
Tracing: Active
Events:
PostProjectEstimatePath:
Type: Api
Properties:
Path: /q/{i}
Method: POST
RestApiId: !Ref MyApi
Layers:
- !Ref ApiSharedLayer
Environment:
Variables:
POWERTOOLS_SERVICE_NAME: PowertoolsHelloWorld
POWERTOOLS_METRICS_NAMESPACE: Powertools
LOG_LEVEL: INFO
Tags:
LambdaPowertools: python
Policies:
- SQSSendMessagePolicy:
QueueName: !Ref MySQSQueue
메시지 클래스를 구현합니다.
from pydantic import BaseModel, Field
from typing import List, Optional
class MySqsRequest(BaseModel):
I: Optional[int] = Field(default=None)
app.py에서 큐에 메시지를 보내는 파이썬 코드를 구현합니다.
@app.post("/q/<i>")
@tracer.capture_method
def post_q(i: int) -> dict:
try:
metrics.add_metric(name="PostMessageFunction", unit=MetricUnit.Count, value=1)
queue_url = 'your queue url here'
request = MySqsRequest(I=i)
message = request.model_dump_json()
send_sqs_message(queue_url, message)
return { 'response_code': 200 }
except ValueError as ve:
msg = str(ve)
logger.warning(f"Error processing prequest: {msg}")
return {
'response_code': 400,
'message': msg
}
except Exception as e:
logger.error(f"Error processing request: {str(e)}")
return {
'response_code': 500,
'message': 'Internal server error occurred while processing request'
}
# ----------------------------------------------------------------------------------------------
def get_queue_url() -> str:
queue_name = get_queue_name()
sqs_client = boto3.client("sqs")
response = sqs_client.get_queue_url(
QueueName=queue_name,
)
return response["QueueUrl"]
# ----------------------------------------------------------------------------------------------
def send_sqs_message(queue_url, message):
sqs_client = boto3.client("sqs")
sqs_client.send_message(
QueueUrl=queue_url,
MessageBody=json.dumps(message),
)
# ----------------------------------------------------------------------------------------------
@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_REST)
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
return app.resolve(event, context)
SQS 람다 트리거: SQS에서 메시지 처리하기
template.yaml에 핸들러 설명을 추가합니다.
PostCompleteFunction:
Type: AWS::Serverless::Function
Properties:
Handler: app-sqs.lambda_handler
CodeUri: test
MemorySize: 1024
Description: PostCompleteFunction 함수
Runtime: python3.12
Architectures:
- x86_64
Tracing: Active
Events:
SQSQueueEstimateEvent:
Type: SQS
Properties:
Queue: !Ref MySQSQueueArn
Layers:
- !Ref ApiSharedLayer
Environment:
Variables:
POWERTOOLS_SERVICE_NAME: PowertoolsHelloWorld
POWERTOOLS_METRICS_NAMESPACE: Powertools
LOG_LEVEL: INFO
Tags:
LambdaPowertools: python
Policies:
- SQSPollerPolicy:
QueueName: !Ref MySQSQueue
app-sqs.py에 파이썬 코드를 추가합니다:
tracer = Tracer()
logger = Logger()
metrics = Metrics(namespace="Powertools")
# ----------------------------------------------------------------------------------------------
processor = BatchProcessor(event_type=EventType.SQS)
# ----------------------------------------------------------------------------------------------
@tracer.capture_method
def record_handler(record: SQSRecord):
payload: str = record.json_body
req = TypeAdapter(MySqsRequest).validate_json(payload)
logger.info(f"I: {req.I}")
# ----------------------------------------------------------------------------------------------
@logger.inject_lambda_context
@tracer.capture_lambda_handler
def lambda_handler(event, context: LambdaContext):
return process_partial_response(
event=event,
record_handler=record_handler,
processor=processor,
context=context,
)
배포
sam validate && \
sam build --use-container && \
sam deploy
샘플 호출
$token과 $api 환경 변수를 설정한 후 호출을 수행합니다:
curl -v -H "Authorization: Bearer $token" -X POST ${api}/q/6 -d @test-curl/q_src.json | jq .
다음과 같은 결과를 보게 됩니다:
{
"response_code": 200
}
로그 확인:
sam logs -n PostCompleteFunction --tail
다음과 유사한 내용을 보게 됩니다:
2025/01/16/[$LATEST]5168b560c9eb4ca4871e0ed3e7f0c20d 2025-01-16T02:15:13.842000 {
"level": "INFO",
"location": "record_handler:585",
"message": "I: 6",
"timestamp": "2025-01-16 02:15:13,842+0000",
"service": "PowertoolsHelloWorld",
"cold_start": true,
"function_name": "asdasdads-PostCompleteFunction-qwe123",
"function_memory_size": "1024",
"function_arn": "arn:aws:lambda:ap-southeast-2:123123123:function:asdasdads-PostCompleteFunction-qwe123",
"function_request_id": "f7f469da-5f33-5087-8613-09d832dcedd7",
"xray_trace_id": "1-67886baf-8fefc4f2eb65405758c4d008"
}
좋습니다!
