Implement Async S3KeySensor

[kaxil]

Based on the research in #32 - Async RedshiftOperator

Acceptance Criteria:

• Unit Tests coverage in the PR (80% Code Coverage -- We will need to add CodeCov separately to measure code cov) with all of them passing
• Example DAG using the async Operator that can be used to run Integration tests that are parametrized via Environment variables. Example - https://github.com/apache/airflow/blob/8a03a505e1df0f9de276038c5509135ac569a667/airflow/providers/google/cloud/example_dags/example_bigquery_to_gcs.py#L33-L35
• Add proper docstrings for each of the methods and functions including Example DAG on how it should be used (populate
• Exception Handling in case of errors
• Improve the OSS Docs to make sure it covers the following:
  • Has an example DAG for the sync version
  • How to add a connection via Environment Variable & explain each of the fields. Example - https://airflow.apache.org/docs/apache-airflow-providers-postgres/stable/connections/postgres.html
  • How to use Guide for the Operator - example: https://airflow.apache.org/docs/apache-airflow-providers-postgres/stable/operators/postgres_operator_howto_guide.html

[phanikumv]

This implementation needs to include the Async version of S3Hook

https://github.com/apache/airflow/blob/206cce971da6941e8c1b0d3c4dbf4fa8afe0fba4/airflow/providers/amazon/aws/hooks/s3.py

[sunank200]

Based on the existing sync implementation, S3KeySensor uses S3Hook. S3KeySensorAsync needs the Async implementation for S3HookAsync. aiobotocore can be used for the hook connection. S3KeySensorAsync can defer itself with a trigger tied to the event that should resume it. This frees up the worker to run something else. The trigger is run until it fires, at which point its source task is re-scheduled. The scheduler queues the task to resume on a worker node. Working on this implementation.

[sunank200]

Draft #45

[sunank200]

Done with changes recommended by Daniel. Working on test.