refactor: domain_dynamo vs HTTP app

lambda/app.py

@@ -1,98 +1,35 @@
-from aws_lambda_powertools import Logger, Tracer, Metrics
-from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
-from aws_lambda_powertools.utilities.typing import LambdaContext
-from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
-import boto3
+"""HTTP handlers — Dynamo access lives in domain_dynamo (boundary discipline)."""
+
+from __future__ import annotations
+
 import json
 import uuid
-import os
 from datetime import datetime, timedelta
-from decimal import Decimal
+
+from aws_lambda_powertools import Logger, Metrics, Tracer
+from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
+from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+# Sibling import: Code.from_asset("lambda") loads app + domain_dynamo as top-level.
+# Tests put lambda/ on sys.path the same way (see tests/test_exclusao.py).
+import domain_dynamo as domain
 
 logger = Logger(service="exclusao-cliente-lambda")
 tracer = Tracer()
 metrics = Metrics(namespace="ExclusaoCliente")
 app = APIGatewayHttpResolver()
 
-TABLE_NAME = os.environ.get("TABLE_NAME", "ExclusaoClientes")
-PIX_FEE = Decimal(str(os.environ.get("PIX_FEE", "50.00")))
-LATEST_SK = "LATEST"
-
-dynamodb = boto3.resource("dynamodb")
-table = dynamodb.Table(TABLE_NAME)
-
-
-def _get_latest(cliente_id: str) -> dict | None:
-    """Read the LATEST pointer row for a cliente."""
-    response = table.get_item(Key={"cliente_id": cliente_id, "request_id": LATEST_SK})
-    return response.get("Item")
-
-
-def _put_solicitacao_with_latest(
-    *,
-    cliente_id: str,
-    request_id: str,
-    motivo: str,
-    qr_code_data: str,
-    fee_amount: Decimal,
-    created_at: str,
-    expires_at: str,
-) -> None:
-    """Write history (SK=uuid) then LATEST pointer (SK=LATEST).
-
-    Ordered puts keep the PR small and moto-friendly. TransactWrite can
-    replace this later if we need stronger atomicity.
-    """
-    history = {
-        "cliente_id": cliente_id,
-        "request_id": request_id,
-        "status": "PENDING_PAYMENT",
-        "motivo": motivo,
-        "qr_code_data": qr_code_data,
-        "fee_amount": fee_amount,
-        "created_at": created_at,
-        "expires_at": expires_at,
-    }
-    latest = {
-        "cliente_id": cliente_id,
-        "request_id": LATEST_SK,
-        "active_request_id": request_id,
-        "status": "PENDING_PAYMENT",
-        "motivo": motivo,
-        "qr_code_data": qr_code_data,
-        "fee_amount": fee_amount,
-        "created_at": created_at,
-        "expires_at": expires_at,
-    }
-    table.put_item(Item=history)
-    table.put_item(Item=latest)
-
-
-def _confirm_payment(cliente_id: str, request_id: str) -> None:
-    """Mark history + LATEST as PAID when pointer matches request_id."""
-    ts = datetime.utcnow().isoformat()
-    table.update_item(
-        Key={"cliente_id": cliente_id, "request_id": request_id},
-        UpdateExpression="SET #status = :paid, updated_at = :ts",
-        ExpressionAttributeNames={"#status": "status"},
-        ExpressionAttributeValues={":paid": "PAID", ":ts": ts},
-        ConditionExpression="attribute_exists(cliente_id)",
-    )
-    table.update_item(
-        Key={"cliente_id": cliente_id, "request_id": LATEST_SK},
-        UpdateExpression="SET #status = :paid, updated_at = :ts",
-        ExpressionAttributeNames={"#status": "status"},
-        ExpressionAttributeValues={
-            ":paid": "PAID",
-            ":ts": ts,
-            ":rid": request_id,
-        },
-        ConditionExpression="active_request_id = :rid",
-    )
-
-
-def generate_pix_qr_code(cliente_id: str, amount: Decimal, request_id: str) -> str:
-    return f"pix-emv-stub://{cliente_id}/{amount}/{request_id}"
+# Re-export for moto fixtures / callers that patch app.table
+TABLE_NAME = domain.TABLE_NAME
+PIX_FEE = domain.PIX_FEE
+LATEST_SK = domain.LATEST_SK
+dynamodb = domain.dynamodb
+table = domain.table
+_get_latest = domain.get_latest
+_put_solicitacao_with_latest = domain.put_solicitacao_with_latest
+_confirm_payment = domain.confirm_payment
+generate_pix_qr_code = domain.generate_pix_qr_code
 
 
 @app.post("/solicitar-exclusao-cliente")
@@ -108,7 +45,7 @@ def solicitar_exclusao(event: APIGatewayProxyEvent):
             "body": json.dumps({"error": "cliente_id is required"}),
         }
 
-    existing = _get_latest(cliente_id)
+    existing = domain.get_latest(cliente_id)
     if existing and existing.get("status") in ("PENDING_PAYMENT", "PAID"):
         return {
             "statusCode": 200,
@@ -124,14 +61,14 @@ def solicitar_exclusao(event: APIGatewayProxyEvent):
     request_id = str(uuid.uuid4())
     timestamp = datetime.utcnow().isoformat()
     expires_at = (datetime.utcnow() + timedelta(days=7)).isoformat()
-    qr_code_data = generate_pix_qr_code(cliente_id, PIX_FEE, request_id)
+    qr_code_data = domain.generate_pix_qr_code(cliente_id, domain.PIX_FEE, request_id)
 
-    _put_solicitacao_with_latest(
+    domain.put_solicitacao_with_latest(
         cliente_id=cliente_id,
         request_id=request_id,
         motivo=motivo,
         qr_code_data=qr_code_data,
-        fee_amount=PIX_FEE,
+        fee_amount=domain.PIX_FEE,
         created_at=timestamp,
         expires_at=expires_at,
     )
@@ -146,7 +83,7 @@ def solicitar_exclusao(event: APIGatewayProxyEvent):
                 "message": "Solicitação registrada",
                 "request_id": request_id,
                 "qr_code": qr_code_data,
-                "fee": float(PIX_FEE),
+                "fee": float(domain.PIX_FEE),
             }
         ),
     }
@@ -156,7 +93,7 @@ def solicitar_exclusao(event: APIGatewayProxyEvent):
 @tracer.capture_method
 def health_check():
     try:
-        table.meta.client.describe_table(TableName=TABLE_NAME)
+        domain.table.meta.client.describe_table(TableName=domain.TABLE_NAME)
         return {"statusCode": 200, "body": json.dumps({"status": "healthy"})}
     except Exception as e:
         return {
@@ -169,7 +106,7 @@ def health_check():
 @tracer.capture_method
 def get_status(cliente_id: str):
     try:
-        item = _get_latest(cliente_id)
+        item = domain.get_latest(cliente_id)
         if item:
             return {
                 "statusCode": 200,
@@ -200,7 +137,7 @@ def confirmar_pagamento(event: APIGatewayProxyEvent):
             ),
         }
     try:
-        _confirm_payment(cliente_id, request_id)
+        domain.confirm_payment(cliente_id, request_id)
         metrics.add_metric(name="exclusao_paga", unit="Count", value=1)
         return {
             "statusCode": 200,

lambda/domain_dynamo.py

@@ -0,0 +1,89 @@
+"""Dynamo access — LATEST pointer (ADR-001). HTTP stays in app.py."""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime
+from decimal import Decimal
+
+import boto3
+
+TABLE_NAME = os.environ.get("TABLE_NAME", "ExclusaoClientes")
+PIX_FEE = Decimal(str(os.environ.get("PIX_FEE", "50.00")))
+LATEST_SK = "LATEST"
+
+dynamodb = boto3.resource("dynamodb")
+table = dynamodb.Table(TABLE_NAME)
+
+
+def get_latest(cliente_id: str) -> dict | None:
+    """Read the LATEST pointer row for a cliente."""
+    response = table.get_item(Key={"cliente_id": cliente_id, "request_id": LATEST_SK})
+    return response.get("Item")
+
+
+def put_solicitacao_with_latest(
+    *,
+    cliente_id: str,
+    request_id: str,
+    motivo: str,
+    qr_code_data: str,
+    fee_amount: Decimal,
+    created_at: str,
+    expires_at: str,
+) -> None:
+    """Write history (SK=uuid) then LATEST pointer (SK=LATEST).
+
+    Ordered puts keep the PR small and moto-friendly. TransactWrite can
+    replace this later if we need stronger atomicity.
+    """
+    history = {
+        "cliente_id": cliente_id,
+        "request_id": request_id,
+        "status": "PENDING_PAYMENT",
+        "motivo": motivo,
+        "qr_code_data": qr_code_data,
+        "fee_amount": fee_amount,
+        "created_at": created_at,
+        "expires_at": expires_at,
+    }
+    latest = {
+        "cliente_id": cliente_id,
+        "request_id": LATEST_SK,
+        "active_request_id": request_id,
+        "status": "PENDING_PAYMENT",
+        "motivo": motivo,
+        "qr_code_data": qr_code_data,
+        "fee_amount": fee_amount,
+        "created_at": created_at,
+        "expires_at": expires_at,
+    }
+    table.put_item(Item=history)
+    table.put_item(Item=latest)
+
+
+def confirm_payment(cliente_id: str, request_id: str) -> None:
+    """Mark history + LATEST as PAID when pointer matches request_id."""
+    ts = datetime.utcnow().isoformat()
+    table.update_item(
+        Key={"cliente_id": cliente_id, "request_id": request_id},
+        UpdateExpression="SET #status = :paid, updated_at = :ts",
+        ExpressionAttributeNames={"#status": "status"},
+        ExpressionAttributeValues={":paid": "PAID", ":ts": ts},
+        ConditionExpression="attribute_exists(cliente_id)",
+    )
+    table.update_item(
+        Key={"cliente_id": cliente_id, "request_id": LATEST_SK},
+        UpdateExpression="SET #status = :paid, updated_at = :ts",
+        ExpressionAttributeNames={"#status": "status"},
+        ExpressionAttributeValues={
+            ":paid": "PAID",
+            ":ts": ts,
+            ":rid": request_id,
+        },
+        ConditionExpression="active_request_id = :rid",
+    )
+
+
+def generate_pix_qr_code(cliente_id: str, amount: Decimal, request_id: str) -> str:
+    return f"pix-emv-stub://{cliente_id}/{amount}/{request_id}"

tests/test_exclusao.py

@@ -4,7 +4,8 @@
     pip install -r requirements.txt
     PYTHONPATH=. pytest tests/test_exclusao.py -v
 
-Note: package dir is named `lambda` (reserved word) — import via importlib.
+Import style matches Lambda Code.from_asset("lambda"): app + domain_dynamo
+as top-level modules (lambda/ on sys.path).
 """
 
 import importlib
@@ -19,7 +20,9 @@
 from moto import mock_aws
 
 ROOT = Path(__file__).resolve().parents[1]
-sys.path.insert(0, str(ROOT))
+LAMBDA_DIR = ROOT / "lambda"
+# Same layout as the Lambda zip root
+sys.path.insert(0, str(LAMBDA_DIR))
 
 os.environ["TABLE_NAME"] = "ExclusaoClientes"
 os.environ["PIX_FEE"] = "50.00"
@@ -31,7 +34,11 @@
 
 
 def _app():
-    return importlib.import_module("lambda.app")
+    return importlib.import_module("app")
+
+
+def _domain():
+    return importlib.import_module("domain_dynamo")
 
 
 @pytest.fixture
@@ -51,11 +58,19 @@ def dynamodb_table():
             BillingMode="PAY_PER_REQUEST",
         )
 
+        domain = _domain()
+        domain.TABLE_NAME = "ExclusaoClientes"
+        domain.table = tbl
+        domain.dynamodb = ddb
+
         app_module = _app()
-        app_module.TABLE_NAME = "ExclusaoClientes"
-        app_module.table = tbl
-        app_module.dynamodb = ddb
-        app_module._ddb_client = ddb.meta.client
+        app_module.TABLE_NAME = domain.TABLE_NAME
+        app_module.table = domain.table
+        app_module.dynamodb = domain.dynamodb
+        app_module.LATEST_SK = domain.LATEST_SK
+        app_module._get_latest = domain.get_latest
+        app_module._put_solicitacao_with_latest = domain.put_solicitacao_with_latest
+        app_module._confirm_payment = domain.confirm_payment
 
         yield tbl