Update S3KeySensor docs

Author: tuzonghua

providers/amazon/docs/operators/s3/s3.rst

@@ -204,8 +204,8 @@ Wait on an Amazon S3 key
 To wait for one or multiple keys to be present in an Amazon S3 bucket you can use
 :class:`~airflow.providers.amazon.aws.sensors.s3.S3KeySensor`.
 For each key, it calls
-`head_object <https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.head_object>`__
-API (or `list_objects_v2 <https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.list_objects_v2>`__
+`head_object <https://docs.aws.amazon.com/boto3/latest/reference/services/s3/client/head_object.html>`__
+API (or `list_objects_v2 <https://docs.aws.amazon.com/boto3/latest/reference/services/s3/client/list_objects_v2.html>`__
 API if ``wildcard_match`` is ``True``) to check whether it is present or not.
 Please keep in mind, especially when used to check a large volume of keys, that it makes one API call per key.
 
@@ -217,6 +217,7 @@ To check one file:
     :start-after: [START howto_sensor_s3_key_single_key]
     :end-before: [END howto_sensor_s3_key_single_key]
 
+
 To check multiple files:
 
 .. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_s3.py
@@ -225,6 +226,7 @@ To check multiple files:
     :start-after: [START howto_sensor_s3_key_multiple_keys]
     :end-before: [END howto_sensor_s3_key_multiple_keys]
 
+
 To check a file with regular expression:
 
 .. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_s3.py
@@ -233,6 +235,7 @@ To check a file with regular expression:
     :start-after: [START howto_sensor_s3_key_regex]
     :end-before: [END howto_sensor_s3_key_regex]
 
+
 To check with an additional custom check you can define a function which receives a list of matched S3 object
 attributes and returns a boolean:
 
@@ -241,11 +244,13 @@ attributes and returns a boolean:
 
 This function is called for each key passed as parameter in ``bucket_key``.
 The reason why the parameter of this function is a list of objects is when ``wildcard_match`` is ``True``,
-multiple files can match one key. The list of matched S3 object attributes contain only the size and is this format:
+multiple files can match one key. The list of matched S3 object attributes contains the associated S3 key as well as
+any valid keys in the ``head_object`` response. If no ``metadata_keys`` are given, it defaults to the size
+and is in this format:
 
 .. code-block:: python
 
-    [{"Size": int}]
+    [{"Key": str, "Size": int}]
 
 .. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_s3.py
     :language: python
@@ -259,6 +264,14 @@ multiple files can match one key. The list of matched S3 object attributes conta
     :start-after: [START howto_sensor_s3_key_function]
     :end-before: [END howto_sensor_s3_key_function]
 
+To filter by S3 key:
+
+.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_s3.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_sensor_s3_key_function_filter_definition]
+    :end-before: [END howto_sensor_s3_key_function_filter_definition]
+
 You can also run this operator in deferrable mode by setting the parameter ``deferrable`` to True.
 This will lead to efficient utilization of Airflow workers as polling for job status happens on
 the triggerer asynchronously. Note that this will need triggerer to be available on your Airflow deployment.

providers/amazon/src/airflow/providers/amazon/aws/sensors/s3.py

@@ -66,6 +66,7 @@ def check_fn(files: List, **kwargs) -> bool:
     :param deferrable: Run operator in the deferrable mode
     :param use_regex: whether to use regex to check bucket
     :param metadata_keys: List of head_object attributes to gather and send to ``check_fn``.
+        Contains the associated S3 key along with list of given attributes.
         Acceptable values: Any top level attribute returned by s3.head_object. Specify * to return
         all available attributes.
         Default value: "Size".
@@ -113,7 +114,7 @@ def _check_key(self, key, context: Context):
         """
         Set variable `files` which contains a list of dict which contains attributes defined by the user
         Format: [{
-            'Size': int
+            'Key': str, 'Size': int
         }]
         """
         if self.wildcard_match:

providers/amazon/tests/system/amazon/aws/example_s3.py

@@ -88,6 +88,21 @@ def check_fn(files: list, **kwargs) -> bool:
 
     # [END howto_sensor_s3_key_function_definition]
 
+    # [START howto_sensor_s3_key_function_filter_definition]
+    def check_fn_with_filter(files: list, **kwargs) -> bool:
+        """
+        Example of custom check: check if one file is bigger than ``20 bytes``
+
+        :param files: List of S3 object attributes.
+        :return: true if the criteria is met
+        """
+        for f in files:
+            if "hadoop" in f.get("Key", ""):
+                return f.get("Size", 0) > 20
+        return True
+
+    # [END howto_sensor_s3_key_function_filter_definition]
+
     # [START howto_operator_s3_create_bucket]
     create_bucket = S3CreateBucketOperator(
         task_id="create_bucket",