add new S3MultiKeySensor

[riastrad]

Over at @sofarocean, we found that the existing S3 sensors did not support one critical use case we had. Specifically, we needed to explicitly check for the presence of multiple S3 keys before continuing on to downstream tasks. If any of the objects were missing, downstream tasks would fail so we needed a sensor that would only resolve to True if all keys were accounted for.

Here is the proposed API:

task = S3MultiKeySensor(
  task_id='task_id',
  bucket_name='bucket'
  bucket_keys=[
    'key1',
    'key2',
    'key3'
  ]
)

I'm sharing the solution we came up with in case it might prove useful to others with similar use cases. I don't have a clear picture of the broader community's preferred dev ergonomics, so if the code needs to be made more robust or if the Airflow team would prefer that I also add support for multiple buckets, full s3:// urls, etc. please just let me know and I'm happy to update this PR.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[o-nikolas]

Hey @riastrad!

Thanks for the contribution :)

Two notes:

1. New operators should come along with example dags and docs. You may be able to update an existing example dag/doc with this new functionality. You can see some examples for S3 here and here.
2. I'd like to hear more about your use case. How many keys are you waiting on? If it's only a few there's nothing stopping you from kicking off several sensors in parallel waiting on keys and then have a downstream task with the trigger rule all_success (implied in chain example): chain(start_task, [wait_for_key_1, wait_for_key_2, wait_for_key_3], end_task)
   This will create a dag like this, where end_task will only run after all upstream tasks have completed:
   

Does this satisfy your usecase, or do you still think this net-new Operator is required?

[eladkal]

I think the use case of this sensor will be covered fully with #22737

[riastrad]

Thank you both for the quick responses!

@o-nikolas:

1. Thanks for heads up about doc updates. If we decide to move forward with this PR then I'll be sure to add those.
2. Our use case currently only requires 3 checks, but might need to expand to more in the future. We had considered using multiple sensors with a downstream trigger rule like you described, but our DAGs are already quite complicated and a custom sensor that only represented one task felt like the cleaner solution in this case.

@eladkal thank you for flagging #22737! I didn't see that when I initially skimmed PRs in this repo. A quick look at those changes and I agree that they should satisfy our use case quite well if/when they land in main.

I'm definitely happy to close this PR out in favor of #22737, since that appears to be a more robust solution 👍

[vincbeck]

Exactly like @eladkal mentioned, once my PR #22737 merged, S3KeySizeSensor will be able to support multiple files

[vincbeck]

Happy to see there is a need for such use case :)

[eladkal]

So I'm closing this PR in favor of #22737
@riastrad it would be great if you can also review #22737 as you are going to use it so your perspective can be very useful.