Add utils for ec_vpc_peer* modules

changelogs/fragments/20240924-ec2-utils.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - module_utils/ec2 - add utils for the ec2_vpc_peer* modules (https://github.com/ansible-collections/amazon.aws/pull/2303).

plugins/module_utils/ec2.py

@@ -318,6 +318,51 @@ def disassociate_vpc_cidr_block(client, association_id: str) -> Dict[str, Any]:
     return client.disassociate_vpc_cidr_block(AssociationId=association_id)
 
 
+# EC2 VPC Peering Connection
+class EC2VpcPeeringErrorHandler(AWSErrorHandler):
+    _CUSTOM_EXCEPTION = AnsibleEC2Error
+
+    @classmethod
+    def _is_missing(cls):
+        return is_boto3_error_code("InvalidVpcPeeringConnectionID.NotFound")
+
+
+@EC2VpcPeeringErrorHandler.list_error_handler("describe vpc peering", [])
+@AWSRetry.jittered_backoff()
+def describe_vpc_peering_connections(client, **params: Dict[str, Any]) -> List[Dict[str, Any]]:
+    paginator = client.get_paginator("describe_vpc_peering_connections")
+    return paginator.paginate(**params).build_full_result()["VpcPeeringConnections"]
+
+
+@EC2VpcSubnetErrorHandler.common_error_handler("create vpc peering")
+@AWSRetry.jittered_backoff()
+def create_vpc_peering_connection(
+    client, **params: Dict[str, Union[str, bool, int, EC2TagSpecifications]]
+) -> Dict[str, Any]:
+    return client.create_vpc_peering_connection(**params)["VpcPeeringConnection"]
+
+
+@EC2VpcSubnetErrorHandler.deletion_error_handler("delete vpc peering")
+@AWSRetry.jittered_backoff()
+def delete_vpc_peering_connection(client, peering_id: str) -> bool:
+    client.delete_vpc_peering_connection(VpcPeeringConnectionId=peering_id)
+    return True
+
+
+@EC2VpcSubnetErrorHandler.deletion_error_handler("accept vpc peering")
+@AWSRetry.jittered_backoff()
+def accept_vpc_peering_connection(client, peering_id: str) -> bool:
+    client.accept_vpc_peering_connection(VpcPeeringConnectionId=peering_id)
+    return True
+
+
+@EC2VpcSubnetErrorHandler.deletion_error_handler("reject vpc peering")
+@AWSRetry.jittered_backoff()
+def reject_vpc_peering_connection(client, peering_id: str) -> bool:
+    client.reject_vpc_peering_connection(VpcPeeringConnectionId=peering_id)
+    return True
+
+
 # EC2 Internet Gateway
 class EC2InternetGatewayErrorHandler(AWSErrorHandler):
     _CUSTOM_EXCEPTION = AnsibleEC2Error