feat(parser): add AWS Transfer Family model

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -109,6 +109,7 @@
 )
 from .sns import SnsModel, SnsNotificationModel, SnsRecordModel
 from .sqs import SqsAttributesModel, SqsModel, SqsMsgAttributeModel, SqsRecordModel
+from .transfer_family import TransferFamily
 from .vpc_lattice import VpcLatticeModel
 from .vpc_latticev2 import VpcLatticeV2Model
 
@@ -179,6 +180,7 @@
     "SqsAttributesModel",
     "S3SqsEventNotificationModel",
     "S3SqsEventNotificationRecordModel",
+    "TransferFamily",
     "APIGatewayProxyEventModel",
     "APIGatewayEventRequestContext",
     "APIGatewayEventAuthorizer",

aws_lambda_powertools/utilities/parser/models/transfer_family.py

@@ -0,0 +1,12 @@
+from typing import Literal, Optional
+
+from pydantic import BaseModel, Field
+from pydantic.networks import IPvAnyAddress
+
+
+class TransferFamily(BaseModel):
+    username: str
+    password: Optional[str] = None
+    protocol: Literal["SFTP", "FTP", "FTPS"]
+    server_id: str = Field(..., alias="serverId")
+    source_ip: IPvAnyAddress = Field(..., alias="sourceIp")

docs/utilities/parser.md

@@ -108,8 +108,8 @@ The example above uses `SqsModel`. Other built-in models can be found below.
 | **ApiGatewayAuthorizerRequest**             | Lambda Event Source payload for Amazon API Gateway Lambda Authorizer with Request     |
 | **APIGatewayProxyEventV2Model**             | Lambda Event Source payload for Amazon API Gateway v2 payload                         |
 | **ApiGatewayAuthorizerRequestV2**           | Lambda Event Source payload for Amazon API Gateway v2 Lambda Authorizer               |
-| **APIGatewayWebSocketMessageEventModel** | Lambda Event Source payload for Amazon API Gateway WebSocket API message body         |
-| **APIGatewayWebSocketConnectEventModel** | Lambda Event Source payload for Amazon API Gateway WebSocket API $connect message     |
+| **APIGatewayWebSocketMessageEventModel**    | Lambda Event Source payload for Amazon API Gateway WebSocket API message body         |
+| **APIGatewayWebSocketConnectEventModel**    | Lambda Event Source payload for Amazon API Gateway WebSocket API $connect message     |
 | **APIGatewayWebSocketDisconnectEventModel** | Lambda Event Source payload for Amazon API Gateway WebSocket API $disconnect message  |
 | **BedrockAgentEventModel**                  | Lambda Event Source payload for Bedrock Agents                                        |
 | **CloudFormationCustomResourceCreateModel** | Lambda Event Source payload for AWS CloudFormation `CREATE` operation                 |
@@ -132,6 +132,7 @@ The example above uses `SqsModel`. Other built-in models can be found below.
 | **SesModel**                                | Lambda Event Source payload for Amazon Simple Email Service                           |
 | **SnsModel**                                | Lambda Event Source payload for Amazon Simple Notification Service                    |
 | **SqsModel**                                | Lambda Event Source payload for Amazon SQS                                            |
+| **TransferFamily**                          | Lambda Event Source payload for AWS Transfer Family custom identity provider          |
 | **VpcLatticeModel**                         | Lambda Event Source payload for Amazon VPC Lattice                                    |
 | **VpcLatticeV2Model**                       | Lambda Event Source payload for Amazon VPC Lattice v2 payload                         |
 

tests/events/TransferFamily.json

@@ -0,0 +1,7 @@
+{
+    "username": "value",
+    "password": "value",
+    "protocol": "SFTP",
+    "serverId": "s-abcd123456",
+    "sourceIp": "192.168.0.100"
+}

tests/unit/parser/_pydantic/test_aws_transfer_family.py

@@ -0,0 +1,25 @@
+from aws_lambda_powertools.utilities.parser.models import TransferFamily
+from tests.functional.utils import load_event
+
+
+def test_aws_transfer_family_model():
+    raw_event = load_event("TransferFamily.json")
+    parsed_event = TransferFamily(**raw_event)
+
+    assert parsed_event.username == raw_event["username"]
+    assert parsed_event.password == raw_event["password"]
+    assert parsed_event.protocol == raw_event["protocol"]
+    assert parsed_event.server_id == raw_event["serverId"]
+    assert str(parsed_event.source_ip) == raw_event["sourceIp"]
+
+
+def test_aws_transfer_family_model_without_password():
+    raw_event = load_event("TransferFamily.json")
+    del raw_event["password"]
+    parsed_event = TransferFamily(**raw_event)
+
+    assert parsed_event.username == raw_event["username"]
+    assert parsed_event.password is None
+    assert parsed_event.protocol == raw_event["protocol"]
+    assert parsed_event.server_id == raw_event["serverId"]
+    assert str(parsed_event.source_ip) == raw_event["sourceIp"]