Struktur proyek Python dalam arsitektur heksagonal menggunakan AWS Lambda

Masuk ke GitHub.

Buat repositori baru. Untuk instruksi, lihat GitHub dokumentasi.

Kloning dan dorong repositori sampel untuk pola ini ke dalam repositori baru di akun Anda.

Instal Puisi.

pip install poetry

Instal paket dari direktori root. Perintah berikut menginstal aplikasi dan paket AWS CDK. Ini juga menginstal paket pengembangan yang diperlukan untuk menjalankan pengujian unit. Semua paket yang diinstal ditempatkan di lingkungan virtual baru.

poetry install

Untuk melihat representasi grafis dari paket yang diinstal, jalankan perintah berikut.

poetry show --tree

Perbarui semua dependensi.

poetry update

Buka shell baru dalam lingkungan virtual yang baru dibuat. Ini berisi semua dependensi yang diinstal.

poetry shell

Perbarui .vscode/settings file.

{
    "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",
} 

Buat .env file di direktori root proyek. Ini memastikan bahwa direktori root proyek disertakan dalam PYTHONPATH sehingga pytest dapat menemukannya dan menemukan semua paket dengan benar.

PYTHONPATH=. 

Pilih penerjemah Python dari lingkungan virtual yang dikelola oleh Puisi.

Jalankan tes dari Test Explorer.

Mulai shell baru dalam lingkungan virtual.

poetry shell

Jalankan pytest perintah dari direktori root.

python -m pytest

Atau Anda dapat menjalankan perintah langsung dari Puisi.

poetry run python -m pytest

Instal AWS CDK v2.

npm install -g aws-cdk

Untuk informasi selengkapnya, lihat dokumentasi AWS CDK.

Bootstrap AWS CDK ke akun dan Wilayah Anda.

cdk bootstrap aws://12345678900/us-east-1 --profile aws-profile-name

Terapkan aplikasi sebagai CloudFormation tumpukan AWS dengan menggunakan profil AWS.

cdk deploy --profile aws-profile-name

Instal Postman sebagai aplikasi mandiri atau ekstensi browser.

Salin URL titik akhir untuk API Gateway. Ini akan dalam format berikut.

https://{api-id}.execute-api.{region}.amazonaws.com/{stage}/{path}

Konfigurasikan tanda tangan AWS di tab otorisasi. Untuk petunjuknya, lihat artikel AWS re:Post tentang mengaktifkan autentikasi IAM untuk API Gateway REST. APIs

Gunakan Postman untuk mengirim permintaan ke titik akhir API Anda.

Buat file Python di app/domain/tests folder dengan menggunakan awalan nama test_ file.

Buat metode pengujian baru untuk menguji logika bisnis baru dengan menggunakan contoh berikut.

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

Buat kelas perintah di app/domain/commands folder. 

Jika fungsionalitasnya baru, buat rintisan untuk pengendali perintah di folder. app/domain/command_handlers

Jalankan unit test untuk melihatnya gagal, karena masih belum ada logika bisnis.

python -m pytest

Menerapkan logika bisnis dalam file handler perintah yang baru dibuat. 

Untuk setiap dependensi yang berinteraksi dengan sistem eksternal, deklarasikan kelas abstrak di folder. 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:
        ...

Perbarui tanda tangan pengendali perintah untuk menerima dependensi yang baru dideklarasikan dengan menggunakan kelas port abstrak sebagai anotasi tipe.

def handle_create_product_command(
    command: create_product_command.CreateProductCommand,
    unit_of_work: unit_of_work.UnitOfWork,
) -> str:
    ...

Perbarui pengujian unit untuk mensimulasikan perilaku semua dependensi yang dideklarasikan untuk pengendali perintah.

    # 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
    )

Perbarui logika pernyataan dalam pengujian untuk memeriksa pemanggilan ketergantungan yang diharapkan.

  # 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")

Jalankan unit test untuk melihatnya berhasil.

python -m pytest

Buat file uji di app/adapters/tests folder dengan menggunakan test_ sebagai awalan nama file.

Gunakan pustaka Moto untuk mengejek layanan AWS.

@pytest.fixture 
def mock_dynamodb(): 
   with moto.mock_dynamodb(): 
    yield boto3.resource("dynamodb", region_name="eu-central-1")

Buat metode pengujian baru untuk uji integrasi adaptor.

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

Buat kelas adaptor di app/adapters folder. Gunakan kelas abstrak dari folder port sebagai kelas dasar.

Jalankan unit test untuk melihatnya gagal, karena masih belum ada logika.

python -m pytest

Menerapkan logika dalam file adaptor yang baru dibuat.

Perbarui pernyataan uji.

# 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,
        }
    )

Jalankan unit test untuk melihatnya berhasil.

python -m pytest

Buat file uji di app/entrypoints/api/tests folder dengan menggunakan test_ sebagai awalan nama file. 

Buat perlengkapan konteks Lambda yang akan digunakan oleh tes untuk memanggil 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()

Buat metode pengujian untuk pemanggilan 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)

Jalankan unit test untuk melihatnya gagal, karena masih belum ada logika.

python -m pytest

Buat fungsi untuk logika bisnis API dan deklarasikan sebagai sumber daya 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."""
...

Menerapkan logika 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()

Jalankan unit test untuk melihatnya berhasil.

python -m pytest