기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.
AWS Lambda를 사용하여 육각형 아키텍처로 Python 프로젝트 구조화
작성자: Furkan Oruc(AWS), Dominik Goby(AWS), Darius Kunce(AWS), Michal Ploski(AWS)
환경: PoC 또는 파일럿 | 기술: 소프트웨어 개발 및 테스트, 클라우드 네이티브, 컨테이너 및 마이크로서비스, 서버리스, 현대화 | AWS 서비스: Amazon DynamoDB; AWS Lambda; Amazon API Gateway |
요약
이 패턴은 AWS Lambda를 사용하여 Python 프로젝트를 육각형 아키텍처로 구성하는 방법을 보여줍니다. 이 패턴은 AWS Cloud Development Kit(AWS CDK)를 코드형 인프라(IaC) 도구로, Amazon API Gateway를 REST API로, Amazon DynamoDB를 지속성 계층으로 사용합니다. 육각형 아키텍처는 도메인 기반 설계 원칙을 따릅니다. 육각형 아키텍처에서 소프트웨어는 도메인, 포트, 어댑터라는 세 가지 구성 요소로 구성됩니다. 육각형 아키텍처와 그 이점에 대한 자세한 내용은 AWS 기반 육각형 아키텍처 구축 가이드를 참조하세요.
사전 조건 및 제한 사항
사전 조건
활성 상태의 AWS 계정.
Python 경험
AWS Lambda, AWS CDK, Amazon API Gateway 및 DynamoDB에 대한 지식
GitHub 계정 (가입 지침 참조)
Git(설치 지침 참조)
변경하고 코드를 푸시하기 위한 코드 편집기 GitHub (예: AWS Cloud9, 비주얼 스튜디오 코드 또는) JetBrains PyCharm
Docker가 설치되고 Docker 데몬이 가동되어 실행 중입니다.
제품 버전
아키텍처
대상 기술 스택
대상 기술 스택은 API Gateway, Lambda 및 DynamoDB를 사용하는 Python 서비스로 구성됩니다. 이 서비스는 DynamoDB 어댑터를 사용하여 데이터를 유지합니다. Lambda를 진입점으로 사용하는 함수를 제공합니다. 이 서비스는 Amazon API Gateway를 사용하여 REST API를 노출합니다. API는 AWS Identity and Access Management(IAM)를 클라이언트 인증에 사용합니다.
대상 아키텍처
구현을 설명하기 위해 이 패턴은 서버리스 대상 아키텍처를 배포합니다. 클라이언트는 API Gateway 엔드포인트에 요청을 보낼 수 있습니다. API Gateway는 육각형 아키텍처 패턴을 구현하는 대상 Lambda 함수에 요청을 전달합니다. Lambda 함수는 DynamoDB 테이블에서 생성, 읽기, 업데이트 및 삭제(CRUD) 작업을 수행합니다.
중요: 이 패턴은 PoC 환경에서 테스트되었습니다. 아키텍처를 프로덕션 환경에 배포하기 전에 보안 검토를 수행하여 위협 모델을 식별하고 보안 코드 베이스를 만들어야 합니다. |
|---|
API는 제품 엔티티에 대한 다섯 가지 작업을 지원합니다.
GET /products는 모든 제품을 돌려 보냅니다.
POST /products는 새로운 제품을 생성합니다.
GET /products/{id}는 특정 제품을 돌려 보냅니다.
PUT /products/{id}는 특정 제품을 업데이트합니다.
DELETE /products/{id}는 특정 제품을 삭제합니다.
다음 폴더 구조를 사용하여 육각형 아키텍처 패턴을 따르도록 프로젝트를 구성할 수 있습니다.
app/ # application code
|--- adapters/ # implementation of the ports defined in the domain
|--- tests/ # adapter unit tests
|--- entrypoints/ # primary adapters, entry points
|--- api/ # api entry point
|--- model/ # api model
|--- tests/ # end to end api tests
|--- domain/ # domain to implement business logic using hexagonal architecture
|--- command_handlers/ # handlers used to execute commands on the domain
|--- commands/ # commands on the domain
|--- events/ # events triggered via the domain
|--- exceptions/ # exceptions defined on the domain
|--- model/ # domain model
|--- ports/ # abstractions used for external communication
|--- tests/ # domain tests
|--- libraries/ # List of 3rd party libraries used by the Lambda function
infra/ # infrastructure code
simple-crud-app.py # AWS CDK v2 app
서비스
Amazon API Gateway는 어떤 규모에서든 개발자가 API를 손쉽게 생성, 게시, 유지 관리, 모니터링 및 보호할 수 있도록 지원하는 완전관리형 서비스입니다.
Amazon DynamoDB는 모든 규모에서 고성능 애플리케이션을 실행하도록 설계된 완전 관리형 서버리스 키 값 NoSQL 데이터베이스입니다.
AWS Lambda는 서버를 프로비저닝하거나 관리하지 않고도 사실상 모든 유형의 애플리케이션이나 백엔드 서비스에 대한 코드를 실행할 수 있는 서버리스 이벤트 기반 컴퓨팅 서비스입니다. 200개 이상의 AWS 서비스와 서비스형 소프트웨어(SaaS) 애플리케이션에서 Lambda 함수를 실행할 수 있으며 사용한 만큼만 비용을 지불하면 됩니다.
도구
Git은 이 패턴의 코드 개발을 위한 버전 제어 시스템으로 사용됩니다.
Python은 이 패턴의 프로그래밍 언어로 사용됩니다. Python은 높은 수준의 데이터 구조와 객체 지향 프로그래밍에 대한 접근 방식을 제공합니다. AWS Lambda는 Python 서비스의 운영을 간소화하는 내장 Python 런타임을 제공합니다.
Visual Studio Code는 이 패턴의 개발 및 테스트를 위한 IDE로 사용됩니다. Python 개발을 지원하는 모든 IDE (예: AWS Cloud9 또는) 를 사용할 수 있습니다. PyCharm
AWS Cloud Development Kit(AWS CDK)는 익숙한 프로그래밍 언어를 사용하여 클라우드 애플리케이션 리소스를 정의할 수 있는 오픈 소스 소프트웨어 개발 프레임워크입니다. 이 패턴은 CDK를 사용하여 클라우드 인프라를 코드로 작성하고 배포합니다.
Poetry는 이 패턴에서 종속성을 관리하는 데 사용됩니다.
Docker는 AWS CDK에서 Lambda 패키지 및 계층을 구축하는 데 사용됩니다.
code
이 패턴의 코드는 GitHub Lambda 육각형 아키텍처 샘플 리포지토리에서 사용할 수 있습니다.
모범 사례
프로덕션 환경에서 이 패턴을 사용하려면 다음 모범 사례를 따릅니다.
이 패턴은 AWS X-Ray를 사용하여 애플리케이션의 진입점, 도메인 및 어댑터를 통해 요청을 추적합니다. AWS X-Ray는 개발자가 병목 현상을 식별하고 높은 지연 시간을 파악하여 애플리케이션 성능을 개선할 수 있도록 지원합니다.
에픽
| 작업 | 설명 | 필요한 기술 |
|---|
자체 리포지토리를 생성합니다. | 에 로그인하십시오. GitHub 새 리포지토리를 생성합니다. 지침은 GitHub 설명서를 참조하십시오. 이 패턴의 샘플 리포지토리를 복제하여 계정의 새 리포지토리에 푸시합니다.
| 앱 개발자 |
종속 항목 설치 | Poetry를 설치합니다. pip install poetry
루트 디렉터리에서 패키지를 설치합니다. 다음 명령은 애플리케이션과 AWS CDK 패키지를 설치합니다. 또한 유닛 테스트를 실행하는 데 필요한 개발 패키지도 설치합니다. 설치된 모든 패키지는 새 가상 환경에 배치됩니다. poetry install
설치된 패키지를 그래픽으로 보려면 다음 명령을 실행합니다. poetry show --tree
모든 종속 항목을 업데이트합니다. poetry update
새로 만든 가상 환경 내에서 새 쉘을 엽니다. 여기에는 설치된 모든 종속성이 포함됩니다. poetry shell
| 앱 개발자 |
IDE를 구성합니다. | Visual Studio Code를 사용하는 것이 좋지만 Python을 지원하는 모든 IDE를 사용할 수 있습니다. 다음 단계는 Visual Studio Code용입니다. .vscode/settings 파일을 업데이트합니다.
{
"python.testing.pytestArgs": [
"app/adapters/tests",
"app/entrypoints/api/tests",
"app/domain/tests"
],
"python.testing.unittestEnabled": false,
"python.testing.pytestEnabled": true,
"python.envFile": "${workspaceFolder}/.env",
}
프로젝트의 루트 디렉터리에서 .env 파일을 생성합니다. 이렇게 하면 프로젝트의 루트 디렉터리가 PYTHONPATH에 포함되어 pytest가 프로젝트를 찾고 모든 패키지를 제대로 검색할 수 있습니다. PYTHONPATH=.
| 앱 개발자 |
유닛 테스트 실행 옵션 1: Visual Studio Code | Poetry에서 관리하는 가상 환경의 Python 인터프리터를 선택합니다. 테스트 탐색기에서 테스트를 실행합니다.
| 앱 개발자 |
유닛 테스트 실행 옵션 2: 쉘 명령어 사용. | 가상 환경 내에서 새 쉘을 시작합니다. poetry shell
루트 디렉터리에서 pytest 명령을 실행합니다. python -m pytest
또는 Poetry에서 직접 명령을 실행할 수도 있습니다. poetry run python -m pytest
| 앱 개발자 |
| 작업 | 설명 | 필요한 기술 |
|---|
임시 보안 인증 정보 요청 | cdk deploy 실행 시 쉘에서 AWS 보안 인증을 사용하려면, AWS IAM Identity Center(AWS Single Sign-On 후속)를 사용하여 임시 보안 인증 정보를 생성합니다. 지침은 AWS IAM Identity Center에서 CLI를 사용하기 위한 단기 보안 인증 정보를 검색하는 방법 블로그 게시물을 참조하세요.
| 앱 개발자, AWS DevOps |
애플리케이션을 배포합니다. | AWS CDK v2를 설치합니다. npm install -g aws-cdk
자세한 내용은 AWS CDK 설명서를 참조하세요. AWS CDK를 계정 및 리전에 부트스트랩합니다. cdk bootstrap aws://12345678900/us-east-1 --profile aws-profile-name
AWS 프로필을 사용하여 애플리케이션을 AWS CloudFormation 스택으로 배포합니다. cdk deploy --profile aws-profile-name
| 앱 개발자, AWS DevOps |
API 테스트 옵션 1: 콘솔 사용. | API Gateway 콘솔을 사용하여 API를 테스트합니다. API 작업 및 요청/응답 메시지에 대한 자세한 내용은 리포지토리에 있는 readme 파일의 API 사용 섹션을 참조하십시오. GitHub | 앱 개발자, AWS DevOps |
API 테스트 옵션 2: Postman 사용. | Postman과 같은 도구를 사용하는 경우. 독립 실행형 애플리케이션 또는 브라우저 확장 프로그램으로 Postman을 설치합니다. API Gateway의 엔드포인트 URL을 복사합니다. 형식은 다음과 같습니다. https://{api-id}.execute-api.{region}.amazonaws.com/{stage}/{path}
권한 부여 탭에서 AWS 서명을 구성합니다. 지침은 API Gateway REST API에 대한 IAM 인증 활성화에 관한 AWS re:Post 문서를 참조하세요. Postman을 사용하여 API 엔드포인트로 요청을 보냅니다.
| 앱 개발자, AWS DevOps |
| 작업 | 설명 | 필요한 기술 |
|---|
비즈니스 도메인에 대한 유닛 테스트를 작성합니다. | test_ 파일 이름 접두사를 사용하여 app/domain/tests 폴더에 Python 파일을 만듭니다.
다음 예제를 사용하여 새 비즈니스 로직을 테스트할 수 있는 새 테스트 방법을 생성합니다. def test_create_product_should_store_in_repository():
# Arrange
command = create_product_command.CreateProductCommand(
name="Test Product",
description="Test Description",
)
# Act
create_product_command_handler.handle_create_product_command(
command=command, unit_of_work=mock_unit_of_work
)
# Assert
app/domain/commands 폴더에 명령 클래스를 생성합니다.
기능이 새로 추가된 경우 app/domain/command_handlers 폴더에 명령 처리기용 스텁을 생성합니다. 아직 비즈니스 로직이 없기 때문에 유닛 테스트를 실행하여 실패하는지 확인합니다. python -m pytest
| 앱 개발자 |
명령 및 명령 처러기를 구현합니다. | 새로 만든 명령 처리기 파일에 비즈니스 로직을 구현합니다. 외부 시스템과 상호 작용하는 모든 종속성에 대해 app/domain/ports 폴더에서 추상 클래스를 선언합니다. class ProductsRepository(ABC):
@abstractmethod
def add(self, product: product.Product) -> None:
...
class UnitOfWork(ABC):
products: ProductsRepository
@abstractmethod
def commit(self) -> None:
...
@abstractmethod
def __enter__(self) -> typing.Any:
...
@abstractmethod
def __exit__(self, *args) -> None:
...
추상 포트 클래스를 형식 주석으로 사용하여 새로 선언된 종속성을 허용하도록 명령 처리기 서명을 업데이트합니다. def handle_create_product_command(
command: create_product_command.CreateProductCommand,
unit_of_work: unit_of_work.UnitOfWork,
) -> str:
...
유닛 테스트를 업데이트하여 명령 처리기에 대해 선언된 모든 종속성의 동작을 시뮬레이션합니다. # Arrange
mock_unit_of_work = unittest.mock.create_autospec(
spec=unit_of_work.UnitOfWork, instance=True
)
mock_unit_of_work.products = unittest.mock.create_autospec(
spec=unit_of_work.ProductsRepository, instance=True
)
테스트에서 어설션 로직을 업데이트하여 예상되는 종속성 호출을 확인합니다. # Assert
mock_unit_of_work.commit.assert_called_once()
product = mock_unit_of_work.products.add.call_args.args[0]
assertpy.assert_that(product.name).is_equal_to("Test Product")
assertpy.assert_that(product.description).is_equal_to("Test Description")
유닛 테스트를 실행하여 성공했는지 확인합니다. python -m pytest
| 앱 개발자 |
보조 어댑터에 대한 통합 테스트를 작성합니다. | test_를 파일 이름 접두사로 사용하여 app/adapters/tests 폴더에 테스트 파일을 생성합니다.
Moto 라이브러리를 사용하여 AWS 서비스를 모의합니다. @pytest.fixture
def mock_dynamodb():
with moto.mock_dynamodb():
yield boto3.resource("dynamodb", region_name="eu-central-1")
어댑터의 통합 테스트를 위한 새 테스트 방법을 생성합니다. def test_add_and_commit_should_store_product(mock_dynamodb):
# Arrange
unit_of_work = dynamodb_unit_of_work.DynamoDBUnitOfWork(
table_name=TEST_TABLE_NAME, dynamodb_client=mock_dynamodb.meta.client
)
current_time = datetime.datetime.now(datetime.timezone.utc).isoformat()
new_product_id = str(uuid.uuid4())
new_product = product.Product(
id=new_product_id,
name="test-name",
description="test-description",
createDate=current_time,
lastUpdateDate=current_time,
)
# Act
with unit_of_work:
unit_of_work.products.add(new_product)
unit_of_work.commit()
# Assert
app/adapters 폴더에 어댑터 클래스를 생성합니다. 포트 폴더의 추상 클래스를 기본 클래스로 사용합니다.
아직 로직이 없으므로 유닛 테스트를 실행하여 실패하는지 확인합니다. python -m pytest
| 앱 개발자 |
보조 어댑터를 구현합니다. | 새로 만든 어댑터 파일에 로직을 구현합니다. 테스트 어설션을 업데이트합니다. # Assert
with unit_of_work_readonly:
product_from_db = unit_of_work_readonly.products.get(new_product_id)
assertpy.assert_that(product_from_db).is_not_none()
assertpy.assert_that(product_from_db.dict()).is_equal_to(
{
"id": new_product_id,
"name": "test-name",
"description": "test-description",
"createDate": current_time,
"lastUpdateDate": current_time,
}
)
유닛 테스트를 실행하여 성공했는지 확인합니다. python -m pytest
| 앱 개발자 |
end-to-end 테스트 작성. | test_를 파일 이름 접두사로 사용하여 app/entrypoints/api/tests 폴더에 테스트 파일을 생성합니다.
테스트에서 Lambda를 호출하는 데 사용할 Lambda 컨텍스트 픽스처를 생성합니다. @pytest.fixture
def lambda_context():
@dataclass
class LambdaContext:
function_name: str = "test"
memory_limit_in_mb: int = 128
invoked_function_arn: str = "arn:aws:lambda:eu-west-1:809313241:function:test"
aws_request_id: str = "52fdfc07-2182-154f-163f-5f0f9a621d72"
return LambdaContext()
API 호출을 위한 테스트 방법을 생성합니다. def test_create_product(lambda_context):
# Arrange
name = "TestName"
description = "Test description"
request = api_model.CreateProductRequest(name=name, description=description)
minimal_event = api_gateway_proxy_event.APIGatewayProxyEvent(
{
"path": "/products",
"httpMethod": "POST",
"requestContext": { # correlation ID
"requestId": "c6af9ac6-7b61-11e6-9a41-93e8deadbeef"
},
"body": json.dumps(request.dict()),
}
)
create_product_func_mock = unittest.mock.create_autospec(
spec=create_product_command_handler.handle_create_product_command
)
handler.create_product_command_handler.handle_create_product_command = (
create_product_func_mock
)
# Act
handler.handler(minimal_event, lambda_context)
아직 로직이 없으므로 유닛 테스트를 실행하여 실패하는지 확인합니다. python -m pytest
| 앱 개발자 |
기본 어댑터를 구현합니다. | API 비즈니스 로직용 함수를 만들고 이를 API 리소스로 선언합니다. @tracer.capture_method
@app.post("/products")
@utils.parse_event(model=api_model.CreateProductRequest, app_context=app)
def create_product(
request: api_model.CreateProductRequest,
) -> api_model.CreateProductResponse:
"""Creates a product."""
...
참고: 표시되는 모든 데코레이터는 AWS Lambda Powertools for Python 라이브러리의 기능입니다. 자세한 내용은 AWS Lambda Powertools for Python 웹사이트를 참조하세요. API 로직을 구현합니다. id=create_product_command_handler.handle_create_product_command(
command=create_product_command.CreateProductCommand(
name=request.name,
description=request.description,
),
unit_of_work=unit_of_work,
)
response = api_model.CreateProductResponse(id=id)
return response.dict()
유닛 테스트를 실행하여 성공했는지 확인합니다. python -m pytest
| 앱 개발자 |
관련 리소스
APG 가이드
AWS 참조
도구
IDE
