Python 람다 함수

Python Lambda Function 제너레이터는 기존 Python 프로젝트에 람다 함수를 추가할 수 있는 기능을 제공합니다.

이 제너레이터는 AWS CDK 인프라 설정이 포함된 새로운 Python 람다 핸들러를 생성합니다. 생성된 백엔드는 서버리스 배포를 위해 AWS Lambda를 사용하며, AWS Lambda Powertools의 Parser를 활용한 선택적 타입 안전성을 제공합니다. 또한 로깅, AWS X-Ray 추적, Cloudwatch 메트릭을 포함한 관측 가능성을 위해 AWS Lambda Powertools를 설정합니다.

사용 방법

람다 함수 생성하기

다음 두 가지 방법으로 새 람다 함수를 생성할 수 있습니다:

VSCode

1. 설치 Nx Console VSCode Plugin 아직 설치하지 않았다면
2. VSCode에서 Nx 콘솔 열기
3. 클릭 Generate (UI) "Common Nx Commands" 섹션에서
4. 검색 @aws/nx-plugin - py#lambda-function
5. 필수 매개변수 입력
6. 클릭 Generate

CLI

pnpm

pnpm nx g @aws/nx-plugin:py#lambda-function

yarn

yarn nx g @aws/nx-plugin:py#lambda-function

npm

npx nx g @aws/nx-plugin:py#lambda-function

bun

bunx nx g @aws/nx-plugin:py#lambda-function

어떤 파일이 변경될지 확인하기 위해 드라이 런을 수행할 수도 있습니다

pnpm

pnpm nx g @aws/nx-plugin:py#lambda-function --dry-run

yarn

yarn nx g @aws/nx-plugin:py#lambda-function --dry-run

npm

npx nx g @aws/nx-plugin:py#lambda-function --dry-run

bun

bunx nx g @aws/nx-plugin:py#lambda-function --dry-run

옵션

매개변수	타입	기본값	설명
project 필수	string	-	The project to add the lambda function to
functionName 필수	string	-	The name of the function to add
functionPath	string	-	The path within the project source directory to add the function to
eventSource 필수	string	Any	Optional event source model to use for the lambda function

제너레이터 출력 결과

제너레이터는 프로젝트에 다음 파일들을 추가합니다:

• 디렉터리<module-name>

  • <lambda-function>.py 함수 구현체

제너레이터는 또한 함수 배포에 사용할 수 있는 CDK 구문을 packages/common/constructs 디렉토리에 생성합니다.

functionPath 옵션이 제공된 경우, 제너레이터는 지정된 경로에 필요한 파일들을 추가합니다:

• 디렉터리<module-name>

  • 디렉터리<custom-path>

    • <function-name>.py 함수 구현체

함수 구현하기

주요 함수 구현체는 <function-name>.py에 위치합니다. 예시:

import os

from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.utilities.parser import event_parser
from aws_lambda_powertools.utilities.parser.models import EventBridgeModel
from aws_lambda_powertools.utilities.typing import LambdaContext

os.environ["POWERTOOLS_METRICS_NAMESPACE"] = "Foo"
os.environ["POWERTOOLS_SERVICE_NAME"] = "Foo"

logger: Logger = Logger()
metrics: Metrics = Metrics()
tracer: Tracer = Tracer()

@tracer.capture_lambda_handler
@metrics.log_metrics
@event_parser(model=EventBridgeModel)
def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    logger.info("Received event", extra={"event": event.model_dump() })
    metrics.add_metric(name="InvocationCount", unit=MetricUnit.Count, value=1)

    try:
        # TODO: 구현 필요
        metrics.add_metric(name="SuccessCount", unit=MetricUnit.Count, value=1)
        # TODO: 필요 시 성공 응답 구현
    except Exception as e:
        logger.exception(e)
        metrics.add_metric(name="ErrorCount", unit=MetricUnit.Count, value=1)
        # TODO: 필요 시 오류 응답 구현

제너레이터는 자동으로 다음 기능들을 설정합니다:

1. 관측 가능성을 위한 AWS Lambda Powertools 통합
2. 메트릭 수집
3. @event_parser를 이용한 타입 안전성

AWS Lambda Powertools를 활용한 관측 가능성

로깅

제너레이터는 AWS Lambda Powertools를 사용해 구조화된 로깅을 구성합니다.

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    logger.info("Received event", extra={"event": event.model_dump()})

팁

디버깅과 모니터링을 용이하게 하기 위해 모든 고유 요청에 상관 ID(correlation id)를 설정하는 것을 권장합니다. 상관 ID 모범 사례는 aws powertools logger 문서를 참조하세요.

로거는 자동으로 다음을 포함합니다:

• 이벤트 요청
• 람다 컨텍스트 정보
• 콜드 스타트 표시기

추적

AWS X-Ray 추적이 자동으로 구성됩니다. 추적에 커스텀 하위 세그먼트를 추가할 수 있습니다:

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    # 새로운 하위 세그먼트 생성
    with tracer.provider.in_subsegment("function-subsegment"):
        # 로직 구현
        return ....

메트릭

CloudWatch 메트릭이 각 요청마다 자동으로 수집됩니다. 커스텀 메트릭 추가 가능:

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    metrics.add_metric(name="NewMetric", unit=MetricUnit.Count, value=1)
    return ...

기본 메트릭 포함 사항:

• 호출 횟수
• 성공/실패 횟수
• 콜드 스타트 메트릭

타입 안전성

람다 함수 생성 시 eventSource를 선택한 경우, 함수는 AWS Lambda Powertools의 @event_parser로 계측됩니다. 예시:

@event_parser(model=EventBridgeModel)
def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    event.detail_type # <- IDE 자동 완성 지원되는 타입 안전 코드

이를 통해 Fast API 작업과 유사한 방식으로 Pydantic을 사용해 데이터 모델을 정의할 수 있습니다.

팁

DynamoDB 스트림이나 EventBridge 이벤트와 같이 이벤트 내에 중첩된 커스텀 데이터가 있는 경우, Envelopes를 사용해 해당 데이터에 대한 타입 안전성을 제공할 수 있습니다.

이벤트 타입을 지정하지 않으려면 eventSource 옵션에서 Any를 선택하면 됩니다.

함수 배포하기

Python Lambda Function 제너레이터는 common/constructs 폴더에 함수 배포를 위한 CDK 구문을 생성합니다. CDK 애플리케이션에서 다음과 같이 사용 가능:

import { MyProjectMyFunction } from ':my-scope/common-constructs';

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    // 스택에 함수 추가
    const fn = new MyProjectMyFunction(this, 'MyFunction');
  }
}

이 구문은 다음을 설정합니다:

1. AWS Lambda 함수
2. CloudWatch 로그 그룹
3. X-Ray 추적 구성
4. CloudWatch 메트릭 네임스페이스

이 함수는 모든 람다 이벤트 소스의 대상으로 사용 가능합니다:

노트

이벤트가 핸들러 함수 내에서 적절히 처리되도록 선택한 eventSource 옵션과 이벤트 소스가 일치하는지 확인하세요.

다음 예제는 Event Bridge를 사용해 일정에 따라 람다 함수를 호출하는 CDK 코드를 보여줍니다:

import { EventPattern, Rule, Schedule } from 'aws-cdk-lib/aws-events';
import { LambdaFunction } from 'aws-cdk-lib/aws-events-targets';
import { MyProjectMyFunction } from ':my-scope/common-constructs';

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    // 스택에 함수 추가
    const fn = new MyProjectMyFunction(this, 'MyFunction');

    // EventBridge 작업에 함수 추가
    const eventRule = new Rule(this, 'MyFunctionScheduleRule', {
      schedule: Schedule.cron({ minute: '15' }),
      targets: [new LambdaFunction(fn)],
    });
  }
}