From 7db910a113e62df8e8b5867adfa245eecfc6c3ba Mon Sep 17 00:00:00 2001
From: Andrey Anshin <Andrey.Anshin@taragol.is>
Date: Thu, 12 Oct 2023 17:09:01 +0400
Subject: [PATCH 1/3] Use base classes for AWS Lambda Operators/Sensors

---
 .../amazon/aws/operators/lambda_function.py   | 37 +++++-----
 .../amazon/aws/sensors/lambda_function.py     | 16 ++---
 .../amazon/aws/triggers/lambda_function.py    |  9 ++-
 .../_partials/generic_parameters.rst          | 69 +++++++++++++++++++
 .../connections/aws.rst                       |  6 ++
 .../operators/lambda.rst                      | 23 ++++++-
 .../aws/operators/test_lambda_function.py     | 43 +++++++++++-
 .../aws/sensors/test_lambda_function.py       | 17 +++++
 8 files changed, 185 insertions(+), 35 deletions(-)
 create mode 100644 docs/apache-airflow-providers-amazon/_partials/generic_parameters.rst

diff --git a/airflow/providers/amazon/aws/operators/lambda_function.py b/airflow/providers/amazon/aws/operators/lambda_function.py
index 61d97ebed3a76..d8c7340bd90b4 100644
--- a/airflow/providers/amazon/aws/operators/lambda_function.py
+++ b/airflow/providers/amazon/aws/operators/lambda_function.py
@@ -19,20 +19,19 @@
 
 import json
 from datetime import timedelta
-from functools import cached_property
 from typing import TYPE_CHECKING, Any, Sequence
 
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException
-from airflow.models import BaseOperator
 from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.operators.base_aws import AwsBaseOperator
 from airflow.providers.amazon.aws.triggers.lambda_function import LambdaCreateFunctionCompleteTrigger
 
 if TYPE_CHECKING:
     from airflow.utils.context import Context
 
 
-class LambdaCreateFunctionOperator(BaseOperator):
+class LambdaCreateFunctionOperator(AwsBaseOperator[LambdaHook]):
     """
     Creates an AWS Lambda function.
 
@@ -62,6 +61,7 @@ class LambdaCreateFunctionOperator(BaseOperator):
     :param aws_conn_id: The AWS connection ID to use
     """
 
+    aws_hook_class = LambdaHook
     template_fields: Sequence[str] = (
         "function_name",
         "runtime",
@@ -69,6 +69,7 @@ class LambdaCreateFunctionOperator(BaseOperator):
         "handler",
         "code",
         "config",
+        *AwsBaseOperator.template_fields,
     )
     ui_color = "#ff7300"
 
@@ -82,12 +83,11 @@ def __init__(
         code: dict,
         description: str | None = None,
         timeout: int | None = None,
-        config: dict = {},
+        config: dict | None = None,
         wait_for_completion: bool = False,
         waiter_max_attempts: int = 60,
         waiter_delay: int = 15,
         deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
-        aws_conn_id: str = "aws_default",
         **kwargs,
     ):
         super().__init__(**kwargs)
@@ -98,16 +98,11 @@ def __init__(
         self.code = code
         self.description = description
         self.timeout = timeout
-        self.config = config
+        self.config = config or {}
         self.wait_for_completion = wait_for_completion
         self.waiter_delay = waiter_delay
         self.waiter_max_attempts = waiter_max_attempts
         self.deferrable = deferrable
-        self.aws_conn_id = aws_conn_id
-
-    @cached_property
-    def hook(self) -> LambdaHook:
-        return LambdaHook(aws_conn_id=self.aws_conn_id)
 
     def execute(self, context: Context):
         self.log.info("Creating AWS Lambda function: %s", self.function_name)
@@ -131,6 +126,9 @@ def execute(self, context: Context):
                     waiter_delay=self.waiter_delay,
                     waiter_max_attempts=self.waiter_max_attempts,
                     aws_conn_id=self.aws_conn_id,
+                    region_name=self.region_name,
+                    verify=self.verify,
+                    botocore_config=self.botocore_config,
                 ),
                 method_name="execute_complete",
                 timeout=timedelta(seconds=self.waiter_max_attempts * self.waiter_delay),
@@ -152,7 +150,7 @@ def execute_complete(self, context: Context, event: dict[str, Any] | None = None
         return event["function_arn"]
 
 
-class LambdaInvokeFunctionOperator(BaseOperator):
+class LambdaInvokeFunctionOperator(AwsBaseOperator[LambdaHook]):
     """
     Invokes an AWS Lambda function.
 
@@ -176,7 +174,14 @@ class LambdaInvokeFunctionOperator(BaseOperator):
     :param aws_conn_id: The AWS connection ID to use
     """
 
-    template_fields: Sequence[str] = ("function_name", "payload", "qualifier", "invocation_type")
+    aws_hook_class = LambdaHook
+    template_fields: Sequence[str] = (
+        "function_name",
+        "payload",
+        "qualifier",
+        "invocation_type",
+        *AwsBaseOperator.template_fields,
+    )
     ui_color = "#ff7300"
 
     def __init__(
@@ -189,7 +194,6 @@ def __init__(
         invocation_type: str | None = None,
         client_context: str | None = None,
         payload: bytes | str | None = None,
-        aws_conn_id: str = "aws_default",
         **kwargs,
     ):
         super().__init__(**kwargs)
@@ -200,11 +204,6 @@ def __init__(
         self.qualifier = qualifier
         self.invocation_type = invocation_type
         self.client_context = client_context
-        self.aws_conn_id = aws_conn_id
-
-    @cached_property
-    def hook(self) -> LambdaHook:
-        return LambdaHook(aws_conn_id=self.aws_conn_id)
 
     def execute(self, context: Context):
         """
diff --git a/airflow/providers/amazon/aws/sensors/lambda_function.py b/airflow/providers/amazon/aws/sensors/lambda_function.py
index 9234bb8e7bc32..a31eccdbc9a91 100644
--- a/airflow/providers/amazon/aws/sensors/lambda_function.py
+++ b/airflow/providers/amazon/aws/sensors/lambda_function.py
@@ -17,20 +17,18 @@
 # under the License.
 from __future__ import annotations
 
-from functools import cached_property
 from typing import TYPE_CHECKING, Any, Sequence
 
+from airflow.exceptions import AirflowException, AirflowSkipException
 from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
+from airflow.providers.amazon.aws.sensors.base_aws import AwsBaseSensor
 from airflow.providers.amazon.aws.utils import trim_none_values
 
 if TYPE_CHECKING:
     from airflow.utils.context import Context
 
-from airflow.exceptions import AirflowException, AirflowSkipException
-from airflow.sensors.base import BaseSensorOperator
-
 
-class LambdaFunctionStateSensor(BaseSensorOperator):
+class LambdaFunctionStateSensor(AwsBaseSensor[LambdaHook]):
     """
     Poll the deployment state of the AWS Lambda function until it reaches a target state.
 
@@ -48,9 +46,11 @@ class LambdaFunctionStateSensor(BaseSensorOperator):
 
     FAILURE_STATES = ("Failed",)
 
+    aws_hook_class = LambdaHook
     template_fields: Sequence[str] = (
         "function_name",
         "qualifier",
+        *AwsBaseSensor.template_fields,
     )
 
     def __init__(
@@ -59,11 +59,9 @@ def __init__(
         function_name: str,
         qualifier: str | None = None,
         target_states: list = ["Active"],
-        aws_conn_id: str = "aws_default",
         **kwargs: Any,
     ) -> None:
         super().__init__(**kwargs)
-        self.aws_conn_id = aws_conn_id
         self.function_name = function_name
         self.qualifier = qualifier
         self.target_states = target_states
@@ -83,7 +81,3 @@ def poke(self, context: Context) -> bool:
             raise AirflowException(message)
 
         return state in self.target_states
-
-    @cached_property
-    def hook(self) -> LambdaHook:
-        return LambdaHook(aws_conn_id=self.aws_conn_id)
diff --git a/airflow/providers/amazon/aws/triggers/lambda_function.py b/airflow/providers/amazon/aws/triggers/lambda_function.py
index 4805d5bb6cf3e..be78fd76709b6 100644
--- a/airflow/providers/amazon/aws/triggers/lambda_function.py
+++ b/airflow/providers/amazon/aws/triggers/lambda_function.py
@@ -44,6 +44,7 @@ def __init__(
         waiter_delay: int = 60,
         waiter_max_attempts: int = 30,
         aws_conn_id: str | None = None,
+        **kwargs,
     ) -> None:
 
         super().__init__(
@@ -62,7 +63,13 @@ def __init__(
             waiter_delay=waiter_delay,
             waiter_max_attempts=waiter_max_attempts,
             aws_conn_id=aws_conn_id,
+            **kwargs,
         )
 
     def hook(self) -> AwsGenericHook:
-        return LambdaHook(aws_conn_id=self.aws_conn_id)
+        return LambdaHook(
+            aws_conn_id=self.aws_conn_id,
+            region_name=self.region_name,
+            verify=self.verify,
+            config=self.botocore_config,
+        )
diff --git a/docs/apache-airflow-providers-amazon/_partials/generic_parameters.rst b/docs/apache-airflow-providers-amazon/_partials/generic_parameters.rst
new file mode 100644
index 0000000000000..1b76094743703
--- /dev/null
+++ b/docs/apache-airflow-providers-amazon/_partials/generic_parameters.rst
@@ -0,0 +1,69 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+
+aws_conn_id
+    Reference to :ref:`Amazon Web Services Connection <howto/connection:aws>` ID.
+    If this parameter is set to ``None`` then the default boto3 behaviour is used without lookup connection.
+    Otherwise use credentials stored into the Connection. Default: ``aws_default``
+
+region_name
+    AWS Region Name. If this parameter is set to ``None`` or omitted then **region_name** from
+    :ref:`AWS Connection Extra Parameter <howto/connection:aws:configuring-the-connection>` will use.
+    Otherwise use specified value instead of connection value. Default: ``None``
+
+verify
+    Whether or not to verify SSL certificates.
+
+    * ``False`` - do not validate SSL certificates.
+    * **path/to/cert/bundle.pem** - A filename of the CA cert bundle to uses. You can specify this argument
+      if you want to use a different CA cert bundle than the one used by botocore.
+
+    If this parameter is set to ``None`` or omitted then **verify** from
+    :ref:`AWS Connection Extra Parameter <howto/connection:aws:configuring-the-connection>` will use.
+    Otherwise use specified value instead of from connection value. Default: ``None``
+
+botocore_config
+    Use provided dictionary to construct a `botocore.config.Config`_.
+    This configuration will able to use for :ref:`howto/connection:aws:avoid-throttling-exceptions`, configure timeouts and etc.
+
+    ..  code-block:: python
+        :caption: Example, for more detail about parameters please have a look `botocore.config.Config`_
+
+        {
+            "signature_version": "unsigned",
+            "s3": {
+                "us_east_1_regional_endpoint": True,
+            },
+            "retries": {
+              "mode": "standard",
+              "max_attempts": 10,
+            },
+            "connect_timeout": 300,
+            "read_timeout": 300,
+            "tcp_keepalive": True,
+        }
+
+    If this parameter is set to ``None`` or omitted then **config_kwargs** from
+    :ref:`AWS Connection Extra Parameter <howto/connection:aws:configuring-the-connection>` will use.
+    Otherwise use specified value instead of connection value. Default: ``None``
+
+    .. note::
+        Empty dictionary ``{}`` uses for construct default botocore Config
+        and will overwrite connection configuration for `botocore.config.Config`_
+
+.. _botocore.config.Config: https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
diff --git a/docs/apache-airflow-providers-amazon/connections/aws.rst b/docs/apache-airflow-providers-amazon/connections/aws.rst
index 0b87e7be93fd8..6dbb8555dcc11 100644
--- a/docs/apache-airflow-providers-amazon/connections/aws.rst
+++ b/docs/apache-airflow-providers-amazon/connections/aws.rst
@@ -127,6 +127,10 @@ Extra (optional)
 
     * ``verify``: Whether or not to verify SSL certificates.
 
+        * ``False`` - do not validate SSL certificates.
+        * **path/to/cert/bundle.pem** - A filename of the CA cert bundle to uses. You can specify this argument
+          if you want to use a different CA cert bundle than the one used by botocore.
+
     The following extra parameters used for specific AWS services:
 
     * ``service_config``: json used to specify configuration/parameters per AWS service / Amazon provider hook,
@@ -396,6 +400,8 @@ provide selected options in the connection's extra field.
     }
 
 
+.. _howto/connection:aws:avoid-throttling-exceptions:
+
 Avoid Throttling exceptions
 ---------------------------
 
diff --git a/docs/apache-airflow-providers-amazon/operators/lambda.rst b/docs/apache-airflow-providers-amazon/operators/lambda.rst
index 01ffb18d822f4..eb052e4b12ea3 100644
--- a/docs/apache-airflow-providers-amazon/operators/lambda.rst
+++ b/docs/apache-airflow-providers-amazon/operators/lambda.rst
@@ -30,6 +30,11 @@ Prerequisite Tasks
 
 .. include:: ../_partials/prerequisite_tasks.rst
 
+Generic Parameters
+------------------
+
+.. include:: ../_partials/generic_parameters.rst
+
 Operators
 ---------
 
@@ -68,10 +73,22 @@ To invoke an AWS lambda function you can use
 
       urllib3.exceptions.ReadTimeoutError: AWSHTTPSConnectionPool(host='lambda.us-east-1.amazonaws.com', port=443): Read timed out. (read timeout=60)
 
-    If you encounter this issue, configure :ref:`howto/connection:aws` to allow ``botocore`` / ``boto3`` to use long connections with
-    timeout or keep-alive settings in the Connection Extras field
+    If you encounter this issue, you need to provide `botocore.config.Config <https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html>`__
+    for use long connections with timeout or keep-alive settings
+
+    By provide **botocore_config** as operator parameter
+
+    ..  code-block:: python
+
+        {
+          "connect_timeout": 900,
+          "read_timeout": 900,
+          "tcp_keepalive": True,
+        }
+
+    Or specify **config_kwargs** in associated :ref:`AWS Connection Extra Parameter <howto/connection:aws:configuring-the-connection>`
 
-    .. code-block:: json
+    ..  code-block:: json
 
         {
           "config_kwargs": {
diff --git a/tests/providers/amazon/aws/operators/test_lambda_function.py b/tests/providers/amazon/aws/operators/test_lambda_function.py
index 583dc1fc54174..723f7584d9937 100644
--- a/tests/providers/amazon/aws/operators/test_lambda_function.py
+++ b/tests/providers/amazon/aws/operators/test_lambda_function.py
@@ -43,6 +43,29 @@
 
 
 class TestLambdaCreateFunctionOperator:
+    def test_init(self):
+        op = LambdaCreateFunctionOperator(
+            task_id="task_test",
+            function_name=FUNCTION_NAME,
+            role=ROLE_ARN,
+            code={
+                "ImageUri": IMAGE_URI,
+            },
+            aws_conn_id="aws_conn_test",
+            region_name="foo-bar-1",
+            verify="/spam/egg.pem",
+            botocore_config={"baz": "qux"},
+        )
+
+        assert op.function_name == FUNCTION_NAME
+        assert op.role == ROLE_ARN
+        assert op.code == {"ImageUri": IMAGE_URI}
+
+        assert op.aws_conn_id == "aws_conn_test"
+        assert op.region_name == "foo-bar-1"
+        assert op.verify == "/spam/egg.pem"
+        assert op.botocore_config == {"baz": "qux"}
+
     @mock.patch.object(LambdaHook, "create_lambda")
     @mock.patch.object(LambdaHook, "conn")
     def test_create_lambda_without_wait_for_completion(self, mock_hook_conn, mock_hook_create_lambda):
@@ -61,7 +84,17 @@ def test_create_lambda_without_wait_for_completion(self, mock_hook_conn, mock_ho
 
     @mock.patch.object(LambdaHook, "create_lambda")
     @mock.patch.object(LambdaHook, "conn")
-    def test_create_lambda_with_wait_for_completion(self, mock_hook_conn, mock_hook_create_lambda):
+    @pytest.mark.parametrize(
+        "op_kwargs",
+        [
+            pytest.param({}, id="no-additional-parameters"),
+            pytest.param(
+                {"region_name": "eu-west-1", "verify": True, "botocore_config": {}},
+                id="additional-parameters",
+            ),
+        ],
+    )
+    def test_create_lambda_with_wait_for_completion(self, mock_hook_conn, mock_hook_create_lambda, op_kwargs):
         operator = LambdaCreateFunctionOperator(
             task_id="task_test",
             function_name=FUNCTION_NAME,
@@ -70,6 +103,8 @@ def test_create_lambda_with_wait_for_completion(self, mock_hook_conn, mock_hook_
                 "ImageUri": IMAGE_URI,
             },
             wait_for_completion=True,
+            aws_conn_id="aws_conn_test",
+            **op_kwargs,
         )
         operator.execute(None)
 
@@ -100,12 +135,18 @@ def test_init(self, payload):
             payload=payload,
             log_type="None",
             aws_conn_id="aws_conn_test",
+            region_name="foo-bar-1",
+            verify="/spam/egg.pem",
+            botocore_config={"baz": "qux"},
         )
         assert lambda_operator.task_id == "test"
         assert lambda_operator.function_name == "test"
         assert lambda_operator.payload == payload
         assert lambda_operator.log_type == "None"
         assert lambda_operator.aws_conn_id == "aws_conn_test"
+        assert lambda_operator.region_name == "foo-bar-1"
+        assert lambda_operator.verify == "/spam/egg.pem"
+        assert lambda_operator.botocore_config == {"baz": "qux"}
 
     @mock.patch.object(LambdaHook, "invoke_lambda")
     @mock.patch.object(LambdaHook, "conn")
diff --git a/tests/providers/amazon/aws/sensors/test_lambda_function.py b/tests/providers/amazon/aws/sensors/test_lambda_function.py
index 652d8996fd159..073d7e4e89d02 100644
--- a/tests/providers/amazon/aws/sensors/test_lambda_function.py
+++ b/tests/providers/amazon/aws/sensors/test_lambda_function.py
@@ -28,6 +28,23 @@
 
 
 class TestLambdaFunctionStateSensor:
+    def test_init(self):
+        op = LambdaFunctionStateSensor(
+            task_id="task_test",
+            function_name=FUNCTION_NAME,
+            aws_conn_id="aws_conn_test",
+            region_name="foo-bar-1",
+            verify="/spam/egg.pem",
+            botocore_config={"baz": "qux"},
+        )
+
+        assert op.function_name == FUNCTION_NAME
+
+        assert op.aws_conn_id == "aws_conn_test"
+        assert op.region_name == "foo-bar-1"
+        assert op.verify == "/spam/egg.pem"
+        assert op.botocore_config == {"baz": "qux"}
+
     @pytest.mark.parametrize(
         "get_function_output, expect_failure, expected",
         [

From 0a1598f2f2d8b893837875d021914ff4ab72891d Mon Sep 17 00:00:00 2001
From: Andrey Anshin <Andrey.Anshin@taragol.is>
Date: Fri, 13 Oct 2023 02:21:45 +0400
Subject: [PATCH 2/3] Apply suggestions from code review

Co-authored-by: Niko Oliveira <onikolas@amazon.com>
---
 .../_partials/generic_parameters.rst          | 29 +++++++++----------
 .../connections/aws.rst                       |  4 +--
 .../operators/lambda.rst                      |  4 +--
 3 files changed, 18 insertions(+), 19 deletions(-)

diff --git a/docs/apache-airflow-providers-amazon/_partials/generic_parameters.rst b/docs/apache-airflow-providers-amazon/_partials/generic_parameters.rst
index 1b76094743703..67d22ca5e7eaf 100644
--- a/docs/apache-airflow-providers-amazon/_partials/generic_parameters.rst
+++ b/docs/apache-airflow-providers-amazon/_partials/generic_parameters.rst
@@ -18,28 +18,28 @@
 
 aws_conn_id
     Reference to :ref:`Amazon Web Services Connection <howto/connection:aws>` ID.
-    If this parameter is set to ``None`` then the default boto3 behaviour is used without lookup connection.
-    Otherwise use credentials stored into the Connection. Default: ``aws_default``
+    If this parameter is set to ``None`` then the default boto3 behaviour is used without a connection lookup.
+    Otherwise use the credentials stored in the Connection. Default: ``aws_default``
 
 region_name
     AWS Region Name. If this parameter is set to ``None`` or omitted then **region_name** from
-    :ref:`AWS Connection Extra Parameter <howto/connection:aws:configuring-the-connection>` will use.
-    Otherwise use specified value instead of connection value. Default: ``None``
+    :ref:`AWS Connection Extra Parameter <howto/connection:aws:configuring-the-connection>` will be used.
+    Otherwise use the specified value instead of the connection value. Default: ``None``
 
 verify
     Whether or not to verify SSL certificates.
 
-    * ``False`` - do not validate SSL certificates.
-    * **path/to/cert/bundle.pem** - A filename of the CA cert bundle to uses. You can specify this argument
+    * ``False`` - Do not validate SSL certificates.
+    * **path/to/cert/bundle.pem** - A filename of the CA cert bundle to use. You can specify this argument
       if you want to use a different CA cert bundle than the one used by botocore.
 
-    If this parameter is set to ``None`` or omitted then **verify** from
-    :ref:`AWS Connection Extra Parameter <howto/connection:aws:configuring-the-connection>` will use.
-    Otherwise use specified value instead of from connection value. Default: ``None``
+    If this parameter is set to ``None`` or is omitted then **verify** from
+    :ref:`AWS Connection Extra Parameter <howto/connection:aws:configuring-the-connection>` will be used.
+    Otherwise use the specified value instead of the connection value. Default: ``None``
 
 botocore_config
-    Use provided dictionary to construct a `botocore.config.Config`_.
-    This configuration will able to use for :ref:`howto/connection:aws:avoid-throttling-exceptions`, configure timeouts and etc.
+    The provided dictionary is used to construct a `botocore.config.Config`_.
+    This configuration can be used to configure :ref:`howto/connection:aws:avoid-throttling-exceptions`, timeouts, etc.
 
     ..  code-block:: python
         :caption: Example, for more detail about parameters please have a look `botocore.config.Config`_
@@ -59,11 +59,10 @@ botocore_config
         }
 
     If this parameter is set to ``None`` or omitted then **config_kwargs** from
-    :ref:`AWS Connection Extra Parameter <howto/connection:aws:configuring-the-connection>` will use.
-    Otherwise use specified value instead of connection value. Default: ``None``
+    :ref:`AWS Connection Extra Parameter <howto/connection:aws:configuring-the-connection>` will be used.
+    Otherwise use the specified value instead of the connection value. Default: ``None``
 
     .. note::
-        Empty dictionary ``{}`` uses for construct default botocore Config
-        and will overwrite connection configuration for `botocore.config.Config`_
+        Specifying an empty dictionary, ``{}``, will overwrite the connection configuration for `botocore.config.Config`_
 
 .. _botocore.config.Config: https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
diff --git a/docs/apache-airflow-providers-amazon/connections/aws.rst b/docs/apache-airflow-providers-amazon/connections/aws.rst
index 6dbb8555dcc11..3354da06350e6 100644
--- a/docs/apache-airflow-providers-amazon/connections/aws.rst
+++ b/docs/apache-airflow-providers-amazon/connections/aws.rst
@@ -127,8 +127,8 @@ Extra (optional)
 
     * ``verify``: Whether or not to verify SSL certificates.
 
-        * ``False`` - do not validate SSL certificates.
-        * **path/to/cert/bundle.pem** - A filename of the CA cert bundle to uses. You can specify this argument
+        * ``False`` - Do not validate SSL certificates.
+        * **path/to/cert/bundle.pem** - A filename of the CA cert bundle to use. You can specify this argument
           if you want to use a different CA cert bundle than the one used by botocore.
 
     The following extra parameters used for specific AWS services:
diff --git a/docs/apache-airflow-providers-amazon/operators/lambda.rst b/docs/apache-airflow-providers-amazon/operators/lambda.rst
index eb052e4b12ea3..ef2576a1f5413 100644
--- a/docs/apache-airflow-providers-amazon/operators/lambda.rst
+++ b/docs/apache-airflow-providers-amazon/operators/lambda.rst
@@ -74,9 +74,9 @@ To invoke an AWS lambda function you can use
       urllib3.exceptions.ReadTimeoutError: AWSHTTPSConnectionPool(host='lambda.us-east-1.amazonaws.com', port=443): Read timed out. (read timeout=60)
 
     If you encounter this issue, you need to provide `botocore.config.Config <https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html>`__
-    for use long connections with timeout or keep-alive settings
+    to use long connections with timeout or keep-alive settings.
 
-    By provide **botocore_config** as operator parameter
+    By providing **botocore_config** as an operator parameter
 
     ..  code-block:: python
 

From 28aef3b617a246255a0f6a71b32371a55807b603 Mon Sep 17 00:00:00 2001
From: Andrey Anshin <Andrey.Anshin@taragol.is>
Date: Tue, 17 Oct 2023 13:20:25 +0400
Subject: [PATCH 3/3] Add aws_template_field helper

---
 .../amazon/aws/operators/base_aws.py          | 13 +++---
 .../amazon/aws/operators/lambda_function.py   |  7 ++--
 .../providers/amazon/aws/sensors/base_aws.py  | 13 +++---
 .../amazon/aws/sensors/lambda_function.py     |  4 +-
 airflow/providers/amazon/aws/utils/mixins.py  | 13 ++++++
 .../providers/amazon/aws/utils/test_mixins.py | 40 +++++++++++++++++++
 6 files changed, 72 insertions(+), 18 deletions(-)
 create mode 100644 tests/providers/amazon/aws/utils/test_mixins.py

diff --git a/airflow/providers/amazon/aws/operators/base_aws.py b/airflow/providers/amazon/aws/operators/base_aws.py
index e6e64b94de1ce..aaa5059afbc60 100644
--- a/airflow/providers/amazon/aws/operators/base_aws.py
+++ b/airflow/providers/amazon/aws/operators/base_aws.py
@@ -20,7 +20,12 @@
 from typing import Sequence
 
 from airflow.models import BaseOperator
-from airflow.providers.amazon.aws.utils.mixins import AwsBaseHookMixin, AwsHookParams, AwsHookType
+from airflow.providers.amazon.aws.utils.mixins import (
+    AwsBaseHookMixin,
+    AwsHookParams,
+    AwsHookType,
+    aws_template_fields,
+)
 
 
 class AwsBaseOperator(BaseOperator, AwsBaseHookMixin[AwsHookType]):
@@ -71,11 +76,7 @@ def execute(self, context):
     :meta private:
     """
 
-    template_fields: Sequence[str] = (
-        "aws_conn_id",
-        "region_name",
-        "botocore_config",
-    )
+    template_fields: Sequence[str] = aws_template_fields()
 
     def __init__(
         self,
diff --git a/airflow/providers/amazon/aws/operators/lambda_function.py b/airflow/providers/amazon/aws/operators/lambda_function.py
index d8c7340bd90b4..5dec3f116f81f 100644
--- a/airflow/providers/amazon/aws/operators/lambda_function.py
+++ b/airflow/providers/amazon/aws/operators/lambda_function.py
@@ -26,6 +26,7 @@
 from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
 from airflow.providers.amazon.aws.operators.base_aws import AwsBaseOperator
 from airflow.providers.amazon.aws.triggers.lambda_function import LambdaCreateFunctionCompleteTrigger
+from airflow.providers.amazon.aws.utils.mixins import aws_template_fields
 
 if TYPE_CHECKING:
     from airflow.utils.context import Context
@@ -62,14 +63,13 @@ class LambdaCreateFunctionOperator(AwsBaseOperator[LambdaHook]):
     """
 
     aws_hook_class = LambdaHook
-    template_fields: Sequence[str] = (
+    template_fields: Sequence[str] = aws_template_fields(
         "function_name",
         "runtime",
         "role",
         "handler",
         "code",
         "config",
-        *AwsBaseOperator.template_fields,
     )
     ui_color = "#ff7300"
 
@@ -175,12 +175,11 @@ class LambdaInvokeFunctionOperator(AwsBaseOperator[LambdaHook]):
     """
 
     aws_hook_class = LambdaHook
-    template_fields: Sequence[str] = (
+    template_fields: Sequence[str] = aws_template_fields(
         "function_name",
         "payload",
         "qualifier",
         "invocation_type",
-        *AwsBaseOperator.template_fields,
     )
     ui_color = "#ff7300"
 
diff --git a/airflow/providers/amazon/aws/sensors/base_aws.py b/airflow/providers/amazon/aws/sensors/base_aws.py
index 3c50ec4576595..08153b0841f79 100644
--- a/airflow/providers/amazon/aws/sensors/base_aws.py
+++ b/airflow/providers/amazon/aws/sensors/base_aws.py
@@ -19,7 +19,12 @@
 
 from typing import Sequence
 
-from airflow.providers.amazon.aws.utils.mixins import AwsBaseHookMixin, AwsHookParams, AwsHookType
+from airflow.providers.amazon.aws.utils.mixins import (
+    AwsBaseHookMixin,
+    AwsHookParams,
+    AwsHookType,
+    aws_template_fields,
+)
 from airflow.sensors.base import BaseSensorOperator
 
 
@@ -70,11 +75,7 @@ def poke(self, context):
     :meta private:
     """
 
-    template_fields: Sequence[str] = (
-        "aws_conn_id",
-        "region_name",
-        "botocore_config",
-    )
+    template_fields: Sequence[str] = aws_template_fields()
 
     def __init__(
         self,
diff --git a/airflow/providers/amazon/aws/sensors/lambda_function.py b/airflow/providers/amazon/aws/sensors/lambda_function.py
index a31eccdbc9a91..57a5973743bd7 100644
--- a/airflow/providers/amazon/aws/sensors/lambda_function.py
+++ b/airflow/providers/amazon/aws/sensors/lambda_function.py
@@ -23,6 +23,7 @@
 from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
 from airflow.providers.amazon.aws.sensors.base_aws import AwsBaseSensor
 from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.providers.amazon.aws.utils.mixins import aws_template_fields
 
 if TYPE_CHECKING:
     from airflow.utils.context import Context
@@ -47,10 +48,9 @@ class LambdaFunctionStateSensor(AwsBaseSensor[LambdaHook]):
     FAILURE_STATES = ("Failed",)
 
     aws_hook_class = LambdaHook
-    template_fields: Sequence[str] = (
+    template_fields: Sequence[str] = aws_template_fields(
         "function_name",
         "qualifier",
-        *AwsBaseSensor.template_fields,
     )
 
     def __init__(
diff --git a/airflow/providers/amazon/aws/utils/mixins.py b/airflow/providers/amazon/aws/utils/mixins.py
index 4f36c79f5712e..a0c59175f8c2b 100644
--- a/airflow/providers/amazon/aws/utils/mixins.py
+++ b/airflow/providers/amazon/aws/utils/mixins.py
@@ -33,6 +33,7 @@
 
 from typing_extensions import final
 
+from airflow.compat.functools import cache
 from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 
@@ -163,3 +164,15 @@ def region(self) -> str | None:
         """Alias for ``region_name``, used for compatibility (deprecated)."""
         warnings.warn(REGION_MSG, AirflowProviderDeprecationWarning, stacklevel=3)
         return self.region_name
+
+
+@cache
+def aws_template_fields(*template_fields: str) -> tuple[str, ...]:
+    """Merge provided template_fields with generic one and return in alphabetical order."""
+    if not all(isinstance(tf, str) for tf in template_fields):
+        msg = (
+            "Expected that all provided arguments are strings, but got "
+            f"{', '.join(map(repr, template_fields))}."
+        )
+        raise TypeError(msg)
+    return tuple(sorted(list({"aws_conn_id", "region_name", "verify"} | set(template_fields))))
diff --git a/tests/providers/amazon/aws/utils/test_mixins.py b/tests/providers/amazon/aws/utils/test_mixins.py
new file mode 100644
index 0000000000000..b127a9c83b086
--- /dev/null
+++ b/tests/providers/amazon/aws/utils/test_mixins.py
@@ -0,0 +1,40 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import pytest
+
+from airflow.providers.amazon.aws.utils.mixins import aws_template_fields
+
+
+def test_aws_template_fields():
+    result = aws_template_fields()
+    assert result == ("aws_conn_id", "region_name", "verify")
+    assert aws_template_fields() is result, "aws_template_fields expect to return cached result"
+
+    assert aws_template_fields("foo", "aws_conn_id", "bar") == (
+        "aws_conn_id",
+        "bar",
+        "foo",
+        "region_name",
+        "verify",
+    )
+
+
+def test_aws_template_fields_incorrect_types():
+    with pytest.raises(TypeError, match="Expected that all provided arguments are strings"):
+        aws_template_fields(1, None, "test")
