Python Lambda Functions

Filter this guidePick generator option values to hide sections that don't apply.

event

Python Lambda Function 생성기는 기존 python 프로젝트에 람다 함수를 추가하는 기능을 제공합니다.

이 생성기는 AWS CDK 또는 Terraform 인프라 설정과 함께 새로운 python 람다 핸들러를 생성합니다. 생성된 백엔드는 서버리스 배포를 위해 AWS Lambda를 사용하며, AWS Lambda Powertools의 Parser를 사용하여 선택적으로 타입 안전성을 제공합니다. 로깅, AWS X-Ray 추적 및 Cloudwatch Metrics를 포함한 관찰성을 위해 AWS Lambda Powertools를 설정합니다.

사용법

Lambda Function 생성

두 가지 방법으로 새로운 Lambda Function을 생성할 수 있습니다:

CLI
VSCode

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

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

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

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

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

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

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

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

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

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

옵션

매개변수	타입	기본값	설명
project 필수	string	-	람다 함수를 추가할 프로젝트
name 필수	string	-	추가할 함수의 이름
functionPath	string	-	함수를 추가할 프로젝트 소스 디렉토리 내의 경로
event	Any \| AlbModel \| APIGatewayProxyEventModel \| ApiGatewayAuthorizerToken \| ApiGatewayAuthorizerRequest \| APIGatewayProxyEventV2Model \| ApiGatewayAuthorizerRequestV2 \| APIGatewayWebSocketMessageEventModel \| APIGatewayWebSocketConnectEventModel \| APIGatewayWebSocketDisconnectEventModel \| BedrockAgentEventModel \| CloudFormationCustomResourceCreateModel \| CloudFormationCustomResourceUpdateModel \| CloudFormationCustomResourceDeleteModel \| CloudWatchLogsModel \| DynamoDBStreamModel \| EventBridgeModel \| IoTCoreThingEvent \| IoTCoreThingTypeEvent \| IoTCoreThingTypeAssociationEvent \| IoTCoreThingGroupEvent \| IoTCoreAddOrRemoveFromThingGroupEvent \| IoTCoreAddOrDeleteFromThingGroupEvent \| KafkaMskEventModel \| KafkaSelfManagedEventModel \| KinesisDataStreamModel \| KinesisFirehoseModel \| KinesisFirehoseSqsModel \| LambdaFunctionUrlModel \| S3BatchOperationModel \| S3EventNotificationEventBridgeModel \| S3Model \| S3ObjectLambdaEvent \| S3SqsEventNotificationModel \| SesModel \| SnsModel \| SqsModel \| TransferFamilyAuthorizer \| VpcLatticeModel \| VpcLatticeV2Model	Any	Lambda 함수에 사용할 선택적 이벤트 소스 모델
infra	lambda \| none	lambda	Lambda 함수를 배포할 인프라 유형
iac	inherit \| cdk \| terraform	inherit	선호하는 IaC 공급자. 기본적으로 초기 선택에서 상속됩니다.
preferInstallDependencies	boolean	true	생성기 실행 후 의존성 설치를 선호할지 여부입니다. 여러 생성기를 일괄 처리할 때 설치를 연기하려면 false로 설정하세요 (후속 생성기가 Nx 프로젝트 그래프를 계산할 수 있도록 필요한 경우 설치는 여전히 실행됩니다). 마지막에 한 번만 설치하세요.

생성기 출력

생성기는 프로젝트에 다음 파일을 추가합니다:

디렉터리<module-name>
- <lambda-function>.py Function implementation

functionPath 옵션이 제공되면, 생성기는 지정된 경로에 필요한 파일을 추가합니다:

디렉터리<module-name>
- 디렉터리<custom-path>
  - <function-name>.py Function implementation

인프라

이 생성기는 선택한 iac를 기반으로 코드형 인프라를 제공하므로, 관련 CDK constructs 또는 Terraform 모듈을 포함하는 packages/common에 프로젝트를 생성합니다.

공통 코드형 인프라 프로젝트는 다음과 같이 구성됩니다:

CDK
Terraform

디렉터리packages/common/constructs
- 디렉터리src
  - 디렉터리app/ 프로젝트/생성기에 특정한 인프라를 위한 Constructs
    …
  - 디렉터리core/ app의 constructs에서 재사용되는 일반 constructs
    …
  - index.ts app에서 constructs를 내보내는 진입점
- project.json 프로젝트 빌드 타겟 및 구성

디렉터리packages/common/terraform
- 디렉터리src
  - 디렉터리app/ 프로젝트/생성기에 특정한 인프라를 위한 Terraform 모듈
    …
  - 디렉터리core/ app의 모듈에서 재사용되는 일반 모듈
    …
- project.json 프로젝트 빌드 타겟 및 구성

생성기는 선택한 iac를 기반으로 함수를 배포하기 위한 코드형 인프라를 생성합니다:

CDK
Terraform

생성기는 함수를 배포하는 데 사용할 수 있는 CDK 구성을 생성하며, 이는 packages/common/constructs 디렉토리에 위치합니다.

생성기는 함수를 배포하는 데 사용할 수 있는 Terraform 모듈을 생성하며, 이는 packages/common/terraform/src/app/lambda-functions/<function-name> 디렉토리에 위치합니다.

아키텍처

배포된 함수는 다음과 같은 아키텍처를 가집니다:

생성기는 함수 자체를 제공하며, 스택에서 이벤트 소스를 연결하여 함수를 호출합니다(예: API Gateway 통합, EventBridge 규칙, S3 알림 또는 SQS 이벤트 소스 매핑).

함수 구현

주요 함수 구현은 <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: Implement
        metrics.add_metric(name="SuccessCount", unit=MetricUnit.Count, value=1)
        # TODO: Implement success response if required
    except Exception as e:
        logger.exception(e)
        metrics.add_metric(name="ErrorCount", unit=MetricUnit.Count, value=1)
        # TODO: Implement error response if required

생성기는 여러 기능을 자동으로 설정합니다:

관찰성을 위한 AWS Lambda Powertools 통합
메트릭 수집
@event_parser를 사용한 타입 안전성

AWS Lambda Powertools를 사용한 관찰성

로깅

생성기는 AWS Lambda Powertools를 사용하여 구조화된 로깅을 구성합니다.

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

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

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

추적

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

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    # Creates a new subsegment
    with tracer.provider.in_subsegment("function-subsegment"):
        # Your logic here
        return ....

메트릭

CloudWatch 메트릭은 각 요청에 대해 자동으로 수집됩니다. 사용자 정의 메트릭을 추가할 수 있습니다:

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

기본 메트릭에는 다음이 포함됩니다:

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

타입 안전성

람다 함수를 생성할 때 event를 선택한 경우, 함수는 @event_parser from AWS Lambda Powertools로 계측됩니다. 예를 들어:

@event_parser(model=EventBridgeModel)
def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    event.detail_type # <- type-safe with IDE autocompletion

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

event에 대해 Any를 선택한 경우, @event_parser 데코레이터가 적용되지 않으며 핸들러는 타입이 지정되지 않은 이벤트 dict를 받습니다. Pydantic 유효성 검사 및 타입이 지정된 필드 액세스를 원하는 경우 특정 event로 다시 생성하세요.

번들링

생성기는 uv를 사용하여 Lambda 배포 패키지를 위한 Python 번들링을 자동으로 구성합니다:

pnpm nx bundle <project-name>

yarn nx bundle <project-name>

npx nx bundle <project-name>

bunx nx bundle <project-name>

이 프로세스는 다음을 사용합니다:

uv export를 사용하여 Python 종속성을 requirements.txt 파일로 내보내기
uv pip install을 사용하여 Lambda 배포를 위한 대상 플랫폼(x86_64-manylinux_2_28)에 종속성 설치

함수 배포

이 생성기는 선택한 iac를 기반으로 CDK 또는 Terraform 인프라 코드를 생성합니다. 이를 사용하여 함수를 배포할 수 있습니다.

CDK
Terraform

이 생성기는 common/constructs 폴더에 함수를 배포하기 위한 CDK 구성을 생성합니다. CDK 애플리케이션에서 다음과 같이 사용할 수 있습니다:

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

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    // Add the function to your stack
    const fn = new MyProjectMyFunction(this, 'MyFunction');
  }
}

다음과 같이 설정됩니다:

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

이 함수는 모든 람다 이벤트 소스의 대상으로 사용할 수 있습니다:

아래 예제는 EventBridge를 사용하여 일정에 따라 람다 함수를 호출하는 CDK 코드를 보여줍니다:

import { 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) {
    // Add the function to your stack
    const fn = new MyProjectMyFunction(this, 'MyFunction');

    // Add the function to an EventBridge scheduled rule
    const eventRule = new Rule(this, 'MyFunctionScheduleRule', {
      schedule: Schedule.cron({ minute: '15' }),
      targets: [new LambdaFunction(fn)],
    });
  }
}

이 생성기는 common/terraform 폴더에 함수를 배포하기 위한 Terraform 모듈을 생성합니다. 배포당 한 번씩 공유 core/asset-bucket 모듈을 인스턴스화하고 해당 bucket_name을 asset_bucket_name 입력으로 전달하세요:

module "asset_bucket" {
  source = "../../common/terraform/src/core/asset-bucket"
}

module "my_project_my_function" {
  source = "../../common/terraform/src/app/lambda-functions/my-project-my-function"

  asset_bucket_name = module.asset_bucket.bucket_name

  env = {
    SOME_VARIABLE = "some value"
  }

  additional_iam_policy_statements = [
    # Add any additional permissions your function needs
  ]
}

다음과 같이 설정됩니다:

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

이 함수는 모든 람다 이벤트 소스의 대상으로 사용할 수 있습니다. 아래 예제는 EventBridge를 사용하여 일정에 따라 람다 함수를 호출하는 Terraform 코드를 보여줍니다:

# EventBridge rule for scheduled execution
resource "aws_cloudwatch_event_rule" "my_function_schedule" {
  name                = "my-function-schedule"
  description         = "Trigger my function every 15 minutes"
  schedule_expression = "cron(15 * * * ? *)"
}

# EventBridge target
resource "aws_cloudwatch_event_target" "lambda_target" {
  rule      = aws_cloudwatch_event_rule.my_function_schedule.name
  target_id = "MyFunctionTarget"
  arn       = module.my_project_my_function.function_arn
}

# Permission for EventBridge to invoke the Lambda function
resource "aws_lambda_permission" "allow_eventbridge" {
  statement_id  = "AllowExecutionFromEventBridge"
  action        = "lambda:InvokeFunction"
  function_name = module.my_project_my_function.function_name
  principal     = "events.amazonaws.com"
  source_arn    = aws_cloudwatch_event_rule.my_function_schedule.arn
}