You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2023/08/12 19:06:14 UTC
[airflow] branch main updated: D401 Support - Providers: Airbyte to Atlassian (Inclusive) (#33354)
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new c645d8e40c D401 Support - Providers: Airbyte to Atlassian (Inclusive) (#33354)
c645d8e40c is described below
commit c645d8e40c167ea1f6c332cdc3ea0ca5a9363205
Author: D. Ferruzzi <fe...@amazon.com>
AuthorDate: Sat Aug 12 12:06:06 2023 -0700
D401 Support - Providers: Airbyte to Atlassian (Inclusive) (#33354)
---
airflow/providers/airbyte/hooks/airbyte.py | 6 +--
airflow/providers/alibaba/cloud/hooks/oss.py | 12 ++---
.../alibaba/cloud/log/oss_task_handler.py | 4 +-
airflow/providers/amazon/aws/hooks/base_aws.py | 10 ++--
airflow/providers/amazon/aws/hooks/batch_client.py | 7 ++-
airflow/providers/amazon/aws/hooks/chime.py | 14 +++--
airflow/providers/amazon/aws/hooks/datasync.py | 4 +-
airflow/providers/amazon/aws/hooks/dms.py | 8 +--
airflow/providers/amazon/aws/hooks/eks.py | 50 ++++++------------
.../aws/hooks/elasticache_replication_group.py | 8 +--
airflow/providers/amazon/aws/hooks/emr.py | 2 +-
airflow/providers/amazon/aws/hooks/glue.py | 20 +++++---
airflow/providers/amazon/aws/hooks/glue_catalog.py | 8 +--
airflow/providers/amazon/aws/hooks/glue_crawler.py | 10 ++--
.../providers/amazon/aws/hooks/lambda_function.py | 2 +-
airflow/providers/amazon/aws/hooks/logs.py | 2 +-
airflow/providers/amazon/aws/hooks/quicksight.py | 4 +-
airflow/providers/amazon/aws/hooks/rds.py | 6 +--
.../providers/amazon/aws/hooks/redshift_data.py | 2 +-
airflow/providers/amazon/aws/hooks/redshift_sql.py | 2 +-
airflow/providers/amazon/aws/hooks/s3.py | 60 +++++++++++-----------
.../providers/amazon/aws/hooks/step_function.py | 2 +-
airflow/providers/amazon/aws/links/emr.py | 2 +-
.../providers/amazon/aws/log/s3_task_handler.py | 2 +-
airflow/providers/amazon/aws/operators/datasync.py | 2 +-
airflow/providers/amazon/aws/operators/dms.py | 10 ++--
airflow/providers/amazon/aws/operators/eks.py | 4 +-
.../providers/amazon/aws/operators/glue_crawler.py | 2 +-
.../amazon/aws/operators/lambda_function.py | 2 +-
.../providers/amazon/aws/operators/sagemaker.py | 24 ++++-----
airflow/providers/amazon/aws/sensors/batch.py | 2 +-
.../amazon/aws/sensors/glue_catalog_partition.py | 4 +-
.../providers/amazon/aws/sensors/glue_crawler.py | 2 +-
airflow/providers/amazon/aws/sensors/s3.py | 4 +-
airflow/providers/amazon/aws/sensors/sagemaker.py | 10 ++--
.../amazon/aws/transfers/imap_attachment_to_s3.py | 2 +-
.../providers/amazon/aws/transfers/s3_to_sftp.py | 2 +-
.../providers/amazon/aws/transfers/sftp_to_s3.py | 2 +-
airflow/providers/amazon/aws/triggers/batch.py | 4 +-
airflow/providers/amazon/aws/triggers/ecs.py | 2 +-
airflow/providers/amazon/aws/triggers/eks.py | 4 +-
airflow/providers/amazon/aws/triggers/sagemaker.py | 2 +-
.../providers/amazon/aws/utils/task_log_fetcher.py | 2 +-
airflow/providers/apache/beam/hooks/beam.py | 22 ++++----
airflow/providers/apache/beam/operators/beam.py | 2 +-
airflow/providers/apache/beam/triggers/beam.py | 4 +-
.../providers/apache/cassandra/hooks/cassandra.py | 12 ++---
airflow/providers/apache/drill/hooks/drill.py | 2 +-
.../apache/druid/transfers/hive_to_druid.py | 2 +-
airflow/providers/apache/hdfs/hooks/webhdfs.py | 8 +--
airflow/providers/apache/hive/hooks/hive.py | 24 ++++-----
airflow/providers/apache/hive/macros/hive.py | 6 +--
.../apache/hive/transfers/mssql_to_hive.py | 2 +-
.../apache/hive/transfers/mysql_to_hive.py | 2 +-
airflow/providers/apache/kafka/hooks/base.py | 2 +-
airflow/providers/apache/kafka/hooks/client.py | 2 +-
airflow/providers/apache/kafka/hooks/consume.py | 2 +-
airflow/providers/apache/kafka/hooks/produce.py | 2 +-
airflow/providers/apache/livy/hooks/livy.py | 16 +++---
airflow/providers/apache/livy/operators/livy.py | 2 +-
airflow/providers/apache/livy/triggers/livy.py | 2 +-
airflow/providers/apache/pinot/hooks/pinot.py | 4 +-
.../providers/apache/spark/hooks/spark_submit.py | 8 +--
airflow/providers/apache/sqoop/operators/sqoop.py | 2 +-
airflow/providers/apprise/hooks/apprise.py | 2 +-
airflow/providers/arangodb/hooks/arangodb.py | 4 +-
airflow/providers/asana/hooks/asana.py | 20 ++++----
67 files changed, 236 insertions(+), 251 deletions(-)
diff --git a/airflow/providers/airbyte/hooks/airbyte.py b/airflow/providers/airbyte/hooks/airbyte.py
index 120f8a5142..f0b68e1ad5 100644
--- a/airflow/providers/airbyte/hooks/airbyte.py
+++ b/airflow/providers/airbyte/hooks/airbyte.py
@@ -52,7 +52,7 @@ class AirbyteHook(HttpHook):
def wait_for_job(self, job_id: str | int, wait_seconds: float = 3, timeout: float | None = 3600) -> None:
"""
- Helper method which polls a job to check if it finishes.
+ Poll a job to check if it finishes.
:param job_id: Required. Id of the Airbyte job
:param wait_seconds: Optional. Number of seconds between checks.
@@ -85,7 +85,7 @@ class AirbyteHook(HttpHook):
def submit_sync_connection(self, connection_id: str) -> Any:
"""
- Submits a job to a Airbyte server.
+ Submit a job to a Airbyte server.
:param connection_id: Required. The ConnectionId of the Airbyte Connection.
"""
@@ -97,7 +97,7 @@ class AirbyteHook(HttpHook):
def get_job(self, job_id: int) -> Any:
"""
- Gets the resource representation for a job in Airbyte.
+ Get the resource representation for a job in Airbyte.
:param job_id: Required. Id of the Airbyte job
"""
diff --git a/airflow/providers/alibaba/cloud/hooks/oss.py b/airflow/providers/alibaba/cloud/hooks/oss.py
index c00aea149c..a675eed706 100644
--- a/airflow/providers/alibaba/cloud/hooks/oss.py
+++ b/airflow/providers/alibaba/cloud/hooks/oss.py
@@ -35,7 +35,7 @@ T = TypeVar("T", bound=Callable)
def provide_bucket_name(func: T) -> T:
- """Function decorator that unifies bucket name and key is a key is provided but not a bucket name."""
+ """Unify bucket name and key if a key is provided but not a bucket name."""
function_signature = signature(func)
@wraps(func)
@@ -53,7 +53,7 @@ def provide_bucket_name(func: T) -> T:
def unify_bucket_name_and_key(func: T) -> T:
- """Function decorator that unifies bucket name and key is a key is provided but not a bucket name."""
+ """Unify bucket name and key if a key is provided but not a bucket name."""
function_signature = signature(func)
@wraps(func)
@@ -91,13 +91,13 @@ class OSSHook(BaseHook):
super().__init__(*args, **kwargs)
def get_conn(self) -> Connection:
- """Returns connection for the hook."""
+ """Return connection for the hook."""
return self.oss_conn
@staticmethod
def parse_oss_url(ossurl: str) -> tuple:
"""
- Parses the OSS Url into a bucket name and key.
+ Parse the OSS Url into a bucket name and key.
:param ossurl: The OSS Url to parse.
:return: the parsed bucket name and key
@@ -131,7 +131,7 @@ class OSSHook(BaseHook):
@provide_bucket_name
def get_bucket(self, bucket_name: str | None = None) -> oss2.api.Bucket:
"""
- Returns a oss2.Bucket object.
+ Return a oss2.Bucket object.
:param bucket_name: the name of the bucket
:return: the bucket object to the bucket name.
@@ -144,7 +144,7 @@ class OSSHook(BaseHook):
@unify_bucket_name_and_key
def load_string(self, key: str, content: str, bucket_name: str | None = None) -> None:
"""
- Loads a string to OSS.
+ Load a string to OSS.
:param key: the path of the object
:param content: str to set as content for the key.
diff --git a/airflow/providers/alibaba/cloud/log/oss_task_handler.py b/airflow/providers/alibaba/cloud/log/oss_task_handler.py
index 484ebe4f8e..602183a992 100644
--- a/airflow/providers/alibaba/cloud/log/oss_task_handler.py
+++ b/airflow/providers/alibaba/cloud/log/oss_task_handler.py
@@ -78,7 +78,7 @@ class OSSTaskHandler(FileTaskHandler, LoggingMixin):
)
def set_context(self, ti):
- """This function is used to set the context of the handler."""
+ """Set the context of the handler."""
super().set_context(ti)
# Local location and remote location is needed to open and
# upload local log file to OSS remote storage.
@@ -157,7 +157,7 @@ class OSSTaskHandler(FileTaskHandler, LoggingMixin):
def oss_read(self, remote_log_location, return_error=False):
"""
- Returns the log at the remote_log_location. Returns '' if no logs are found or there is an error.
+ Return the log at the remote_log_location or '' if no logs are found or there is an error.
:param remote_log_location: the log's location in remote storage
:param return_error: if True, returns a string error message if an
diff --git a/airflow/providers/amazon/aws/hooks/base_aws.py b/airflow/providers/amazon/aws/hooks/base_aws.py
index a908882e89..904ed0a385 100644
--- a/airflow/providers/amazon/aws/hooks/base_aws.py
+++ b/airflow/providers/amazon/aws/hooks/base_aws.py
@@ -471,7 +471,7 @@ class AwsGenericHook(BaseHook, Generic[BaseAwsConnection]):
@classmethod
def _get_provider_version(cls) -> str:
- """Checks the Providers Manager for the package version."""
+ """Check the Providers Manager for the package version."""
try:
manager = ProvidersManager()
hook = manager.hooks[cls.conn_type]
@@ -774,7 +774,7 @@ class AwsGenericHook(BaseHook, Generic[BaseAwsConnection]):
@staticmethod
def get_ui_field_behaviour() -> dict[str, Any]:
- """Returns custom UI field behaviour for AWS Connection."""
+ """Return custom UI field behaviour for AWS Connection."""
return {
"hidden_fields": ["host", "schema", "port"],
"relabeling": {
@@ -890,7 +890,7 @@ class AwsGenericHook(BaseHook, Generic[BaseAwsConnection]):
@staticmethod
def _apply_parameters_value(config: dict, waiter_name: str, parameters: dict[str, str] | None) -> dict:
- """Replaces potential jinja templates in acceptors definition."""
+ """Replace potential jinja templates in acceptors definition."""
# only process the waiter we're going to use to not raise errors for missing params for other waiters.
acceptors = config["waiters"][waiter_name]["acceptors"]
for a in acceptors:
@@ -905,7 +905,7 @@ class AwsGenericHook(BaseHook, Generic[BaseAwsConnection]):
return config
def list_waiters(self) -> list[str]:
- """Returns a list containing the names of all waiters for the service, official and custom."""
+ """Return a list containing the names of all waiters for the service, official and custom."""
return [*self._list_official_waiters(), *self._list_custom_waiters()]
def _list_official_waiters(self) -> list[str]:
@@ -944,7 +944,7 @@ class AwsBaseHook(AwsGenericHook[Union[boto3.client, boto3.resource]]):
def resolve_session_factory() -> type[BaseSessionFactory]:
- """Resolves custom SessionFactory class."""
+ """Resolve custom SessionFactory class."""
clazz = conf.getimport("aws", "session_factory", fallback=None)
if not clazz:
return BaseSessionFactory
diff --git a/airflow/providers/amazon/aws/hooks/batch_client.py b/airflow/providers/amazon/aws/hooks/batch_client.py
index 87637be410..74dbef1eac 100644
--- a/airflow/providers/amazon/aws/hooks/batch_client.py
+++ b/airflow/providers/amazon/aws/hooks/batch_client.py
@@ -552,14 +552,11 @@ class BatchClientHook(AwsBaseHook):
@staticmethod
def exponential_delay(tries: int) -> float:
"""
- An exponential back-off delay, with random jitter.
+ Apply an exponential back-off delay, with random jitter.
There is a maximum interval of 10 minutes (with random jitter between 3 and 10 minutes).
This is used in the :py:meth:`.poll_for_job_status` method.
- :param tries: Number of tries
-
-
Examples of behavior:
.. code-block:: python
@@ -589,6 +586,8 @@ class BatchClientHook(AwsBaseHook):
- https://docs.aws.amazon.com/general/latest/gr/api-retries.html
- https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/
+
+ :param tries: Number of tries
"""
max_interval = 600.0 # results in 3 to 10 minute delay
delay = 1 + pow(tries * 0.6, 2)
diff --git a/airflow/providers/amazon/aws/hooks/chime.py b/airflow/providers/amazon/aws/hooks/chime.py
index e2ac1a35a0..e7e37ea7b9 100644
--- a/airflow/providers/amazon/aws/hooks/chime.py
+++ b/airflow/providers/amazon/aws/hooks/chime.py
@@ -74,10 +74,9 @@ class ChimeWebhookHook(HttpHook):
def _build_chime_payload(self, message: str) -> str:
"""
- Builds payload for Chime and ensures messages do not exceed max length allowed.
+ Build payload for Chime and ensures messages do not exceed max length allowed.
- :param message: The message you want to send to your Chime room.
- (max 4096 characters)
+ :param message: The message you want to send to your Chime room. (max 4096 characters)
"""
payload: dict[str, Any] = {}
# We need to make sure that the message does not exceed the max length for Chime
@@ -88,11 +87,10 @@ class ChimeWebhookHook(HttpHook):
return json.dumps(payload)
def send_message(self, message: str) -> None:
- """Execute calling the Chime webhook endpoint.
-
- :param message: The message you want to send to your Chime room.
- (max 4096 characters)
+ """
+ Execute calling the Chime webhook endpoint.
+ :param message: The message you want to send to your Chime room.(max 4096 characters)
"""
chime_payload = self._build_chime_payload(message)
self.run(
@@ -101,7 +99,7 @@ class ChimeWebhookHook(HttpHook):
@classmethod
def get_ui_field_behaviour(cls) -> dict[str, Any]:
- """Returns custom field behaviour to only get what is needed for Chime webhooks to function."""
+ """Return custom field behaviour to only get what is needed for Chime webhooks to function."""
return {
"hidden_fields": ["login", "port", "extra"],
"relabeling": {
diff --git a/airflow/providers/amazon/aws/hooks/datasync.py b/airflow/providers/amazon/aws/hooks/datasync.py
index 32fc762153..841255ed95 100644
--- a/airflow/providers/amazon/aws/hooks/datasync.py
+++ b/airflow/providers/amazon/aws/hooks/datasync.py
@@ -63,7 +63,7 @@ class DataSyncHook(AwsBaseHook):
def create_location(self, location_uri: str, **create_location_kwargs) -> str:
"""
- Creates a new location.
+ Create a new location.
.. seealso::
- :external+boto3:py:meth:`DataSync.Client.create_location_s3`
@@ -174,7 +174,7 @@ class DataSyncHook(AwsBaseHook):
self.get_conn().delete_task(TaskArn=task_arn)
def _refresh_tasks(self) -> None:
- """Refreshes the local list of Tasks."""
+ """Refresh the local list of Tasks."""
tasks = self.get_conn().list_tasks()
self.tasks = tasks["Tasks"]
while "NextToken" in tasks:
diff --git a/airflow/providers/amazon/aws/hooks/dms.py b/airflow/providers/amazon/aws/hooks/dms.py
index 6950edadc8..f4bc29cbe9 100644
--- a/airflow/providers/amazon/aws/hooks/dms.py
+++ b/airflow/providers/amazon/aws/hooks/dms.py
@@ -154,7 +154,7 @@ class DmsHook(AwsBaseHook):
**kwargs,
):
"""
- Starts replication task.
+ Start replication task.
.. seealso::
- :external+boto3:py:meth:`DatabaseMigrationService.Client.start_replication_task`
@@ -172,7 +172,7 @@ class DmsHook(AwsBaseHook):
def stop_replication_task(self, replication_task_arn):
"""
- Stops replication task.
+ Stop replication task.
.. seealso::
- :external+boto3:py:meth:`DatabaseMigrationService.Client.stop_replication_task`
@@ -184,7 +184,7 @@ class DmsHook(AwsBaseHook):
def delete_replication_task(self, replication_task_arn):
"""
- Starts replication task deletion and waits for it to be deleted.
+ Start replication task deletion and waits for it to be deleted.
.. seealso::
- :external+boto3:py:meth:`DatabaseMigrationService.Client.delete_replication_task`
@@ -198,7 +198,7 @@ class DmsHook(AwsBaseHook):
def wait_for_task_status(self, replication_task_arn: str, status: DmsTaskWaiterStatus):
"""
- Waits for replication task to reach status; supported statuses: deleted, ready, running, stopped.
+ Wait for replication task to reach status; supported statuses: deleted, ready, running, stopped.
:param status: Status to wait for
:param replication_task_arn: Replication task ARN
diff --git a/airflow/providers/amazon/aws/hooks/eks.py b/airflow/providers/amazon/aws/hooks/eks.py
index 41cc49c465..f6e3ed83f8 100644
--- a/airflow/providers/amazon/aws/hooks/eks.py
+++ b/airflow/providers/amazon/aws/hooks/eks.py
@@ -102,7 +102,7 @@ class EksHook(AwsBaseHook):
**kwargs,
) -> dict:
"""
- Creates an Amazon EKS control plane.
+ Create an Amazon EKS control plane.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.create_cluster`
@@ -111,7 +111,6 @@ class EksHook(AwsBaseHook):
:param roleArn: The Amazon Resource Name (ARN) of the IAM role that provides permissions
for the Kubernetes control plane to make calls to AWS API operations on your behalf.
:param resourcesVpcConfig: The VPC configuration used by the cluster control plane.
-
:return: Returns descriptive information about the created EKS Cluster.
"""
eks_client = self.conn
@@ -134,7 +133,7 @@ class EksHook(AwsBaseHook):
**kwargs,
) -> dict:
"""
- Creates an Amazon EKS managed node group for an Amazon EKS Cluster.
+ Create an Amazon EKS managed node group for an Amazon EKS Cluster.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.create_nodegroup`
@@ -144,7 +143,6 @@ class EksHook(AwsBaseHook):
:param subnets: The subnets to use for the Auto Scaling group that is created for your nodegroup.
:param nodeRole: The Amazon Resource Name (ARN) of the IAM role to associate with your nodegroup.
:param tags: Optional tags to apply to your nodegroup.
-
:return: Returns descriptive information about the created EKS Managed Nodegroup.
"""
eks_client = self.conn
@@ -182,7 +180,7 @@ class EksHook(AwsBaseHook):
**kwargs,
) -> dict:
"""
- Creates an AWS Fargate profile for an Amazon EKS cluster.
+ Create an AWS Fargate profile for an Amazon EKS cluster.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.create_fargate_profile`
@@ -192,7 +190,6 @@ class EksHook(AwsBaseHook):
:param podExecutionRoleArn: The Amazon Resource Name (ARN) of the pod execution role to
use for pods that match the selectors in the Fargate profile.
:param selectors: The selectors to match for pods to use this Fargate profile.
-
:return: Returns descriptive information about the created Fargate profile.
"""
eks_client = self.conn
@@ -214,13 +211,12 @@ class EksHook(AwsBaseHook):
def delete_cluster(self, name: str) -> dict:
"""
- Deletes the Amazon EKS Cluster control plane.
+ Delete the Amazon EKS Cluster control plane.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.delete_cluster`
:param name: The name of the cluster to delete.
-
:return: Returns descriptive information about the deleted EKS Cluster.
"""
eks_client = self.conn
@@ -232,14 +228,13 @@ class EksHook(AwsBaseHook):
def delete_nodegroup(self, clusterName: str, nodegroupName: str) -> dict:
"""
- Deletes an Amazon EKS managed node group from a specified cluster.
+ Delete an Amazon EKS managed node group from a specified cluster.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.delete_nodegroup`
:param clusterName: The name of the Amazon EKS Cluster that is associated with your nodegroup.
:param nodegroupName: The name of the nodegroup to delete.
-
:return: Returns descriptive information about the deleted EKS Managed Nodegroup.
"""
eks_client = self.conn
@@ -255,14 +250,13 @@ class EksHook(AwsBaseHook):
def delete_fargate_profile(self, clusterName: str, fargateProfileName: str) -> dict:
"""
- Deletes an AWS Fargate profile from a specified Amazon EKS cluster.
+ Delete an AWS Fargate profile from a specified Amazon EKS cluster.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.delete_fargate_profile`
:param clusterName: The name of the Amazon EKS cluster associated with the Fargate profile to delete.
:param fargateProfileName: The name of the Fargate profile to delete.
-
:return: Returns descriptive information about the deleted Fargate profile.
"""
eks_client = self.conn
@@ -280,14 +274,13 @@ class EksHook(AwsBaseHook):
def describe_cluster(self, name: str, verbose: bool = False) -> dict:
"""
- Returns descriptive information about an Amazon EKS Cluster.
+ Return descriptive information about an Amazon EKS Cluster.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.describe_cluster`
:param name: The name of the cluster to describe.
:param verbose: Provides additional logging if set to True. Defaults to False.
-
:return: Returns descriptive information about a specific EKS Cluster.
"""
eks_client = self.conn
@@ -304,7 +297,7 @@ class EksHook(AwsBaseHook):
def describe_nodegroup(self, clusterName: str, nodegroupName: str, verbose: bool = False) -> dict:
"""
- Returns descriptive information about an Amazon EKS managed node group.
+ Return descriptive information about an Amazon EKS managed node group.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.describe_nodegroup`
@@ -312,7 +305,6 @@ class EksHook(AwsBaseHook):
:param clusterName: The name of the Amazon EKS Cluster associated with the nodegroup.
:param nodegroupName: The name of the nodegroup to describe.
:param verbose: Provides additional logging if set to True. Defaults to False.
-
:return: Returns descriptive information about a specific EKS Nodegroup.
"""
eks_client = self.conn
@@ -336,7 +328,7 @@ class EksHook(AwsBaseHook):
self, clusterName: str, fargateProfileName: str, verbose: bool = False
) -> dict:
"""
- Returns descriptive information about an AWS Fargate profile.
+ Return descriptive information about an AWS Fargate profile.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.describe_fargate_profile`
@@ -344,7 +336,6 @@ class EksHook(AwsBaseHook):
:param clusterName: The name of the Amazon EKS Cluster associated with the Fargate profile.
:param fargateProfileName: The name of the Fargate profile to describe.
:param verbose: Provides additional logging if set to True. Defaults to False.
-
:return: Returns descriptive information about an AWS Fargate profile.
"""
eks_client = self.conn
@@ -367,13 +358,12 @@ class EksHook(AwsBaseHook):
def get_cluster_state(self, clusterName: str) -> ClusterStates:
"""
- Returns the current status of a given Amazon EKS Cluster.
+ Return the current status of a given Amazon EKS Cluster.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.describe_cluster`
:param clusterName: The name of the cluster to check.
-
:return: Returns the current status of a given Amazon EKS Cluster.
"""
eks_client = self.conn
@@ -387,14 +377,13 @@ class EksHook(AwsBaseHook):
def get_fargate_profile_state(self, clusterName: str, fargateProfileName: str) -> FargateProfileStates:
"""
- Returns the current status of a given AWS Fargate profile.
+ Return the current status of a given AWS Fargate profile.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.describe_fargate_profile`
:param clusterName: The name of the Amazon EKS Cluster associated with the Fargate profile.
:param fargateProfileName: The name of the Fargate profile to check.
-
:return: Returns the current status of a given AWS Fargate profile.
"""
eks_client = self.conn
@@ -414,14 +403,13 @@ class EksHook(AwsBaseHook):
def get_nodegroup_state(self, clusterName: str, nodegroupName: str) -> NodegroupStates:
"""
- Returns the current status of a given Amazon EKS managed node group.
+ Return the current status of a given Amazon EKS managed node group.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.describe_nodegroup`
:param clusterName: The name of the Amazon EKS Cluster associated with the nodegroup.
:param nodegroupName: The name of the nodegroup to check.
-
:return: Returns the current status of a given Amazon EKS Nodegroup.
"""
eks_client = self.conn
@@ -442,13 +430,12 @@ class EksHook(AwsBaseHook):
verbose: bool = False,
) -> list:
"""
- Lists all Amazon EKS Clusters in your AWS account.
+ List all Amazon EKS Clusters in your AWS account.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.list_clusters`
:param verbose: Provides additional logging if set to True. Defaults to False.
-
:return: A List containing the cluster names.
"""
eks_client = self.conn
@@ -462,14 +449,13 @@ class EksHook(AwsBaseHook):
verbose: bool = False,
) -> list:
"""
- Lists all Amazon EKS managed node groups associated with the specified cluster.
+ List all Amazon EKS managed node groups associated with the specified cluster.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.list_nodegroups`
:param clusterName: The name of the Amazon EKS Cluster containing nodegroups to list.
:param verbose: Provides additional logging if set to True. Defaults to False.
-
:return: A List of nodegroup names within the given cluster.
"""
eks_client = self.conn
@@ -483,14 +469,13 @@ class EksHook(AwsBaseHook):
verbose: bool = False,
) -> list:
"""
- Lists all AWS Fargate profiles associated with the specified cluster.
+ List all AWS Fargate profiles associated with the specified cluster.
.. seealso::
- :external+boto3:py:meth:`EKS.Client.list_fargate_profiles`
:param clusterName: The name of the Amazon EKS Cluster containing Fargate profiles to list.
:param verbose: Provides additional logging if set to True. Defaults to False.
-
:return: A list of Fargate profile names within a given cluster.
"""
eks_client = self.conn
@@ -502,12 +487,11 @@ class EksHook(AwsBaseHook):
def _list_all(self, api_call: Callable, response_key: str, verbose: bool) -> list:
"""
- Repeatedly calls a provided boto3 API Callable and collates the responses into a List.
+ Repeatedly call a provided boto3 API Callable and collates the responses into a List.
:param api_call: The api command to execute.
:param response_key: Which dict key to collect into the final list.
:param verbose: Provides additional logging if set to True. Defaults to False.
-
:return: A List of the combined results of the provided API call.
"""
name_collection: list = []
@@ -532,7 +516,7 @@ class EksHook(AwsBaseHook):
pod_namespace: str | None,
) -> Generator[str, None, None]:
"""
- Writes the kubeconfig file given an EKS Cluster.
+ Write the kubeconfig file given an EKS Cluster.
:param eks_cluster_name: The name of the cluster to generate kubeconfig file for.
:param pod_namespace: The namespace to run within kubernetes.
diff --git a/airflow/providers/amazon/aws/hooks/elasticache_replication_group.py b/airflow/providers/amazon/aws/hooks/elasticache_replication_group.py
index ae66e95352..5dd227d7ee 100644
--- a/airflow/providers/amazon/aws/hooks/elasticache_replication_group.py
+++ b/airflow/providers/amazon/aws/hooks/elasticache_replication_group.py
@@ -62,7 +62,7 @@ class ElastiCacheReplicationGroupHook(AwsBaseHook):
def create_replication_group(self, config: dict) -> dict:
"""
- Creates a Redis (cluster mode disabled) or a Redis (cluster mode enabled) replication group.
+ Create a Redis (cluster mode disabled) or a Redis (cluster mode enabled) replication group.
.. seealso::
- :external+boto3:py:meth:`ElastiCache.Client.create_replication_group`
@@ -74,7 +74,7 @@ class ElastiCacheReplicationGroupHook(AwsBaseHook):
def delete_replication_group(self, replication_group_id: str) -> dict:
"""
- Deletes an existing replication group.
+ Delete an existing replication group.
.. seealso::
- :external+boto3:py:meth:`ElastiCache.Client.delete_replication_group`
@@ -110,7 +110,7 @@ class ElastiCacheReplicationGroupHook(AwsBaseHook):
def is_replication_group_available(self, replication_group_id: str) -> bool:
"""
- Helper for checking if replication group is available or not.
+ Check if replication group is available or not.
:param replication_group_id: ID of replication group to check for availability
:return: True if available else False
@@ -179,7 +179,7 @@ class ElastiCacheReplicationGroupHook(AwsBaseHook):
max_retries: int | None = None,
):
"""
- Helper for deleting a replication group ensuring it is either deleted or can't be deleted.
+ Delete a replication group ensuring it is either deleted or can't be deleted.
:param replication_group_id: ID of replication to delete
:param initial_sleep_time: Initial sleep time in second
diff --git a/airflow/providers/amazon/aws/hooks/emr.py b/airflow/providers/amazon/aws/hooks/emr.py
index 185c7ab13f..54127d98a3 100644
--- a/airflow/providers/amazon/aws/hooks/emr.py
+++ b/airflow/providers/amazon/aws/hooks/emr.py
@@ -207,7 +207,7 @@ class EmrHook(AwsBaseHook):
@staticmethod
def get_ui_field_behaviour() -> dict[str, Any]:
- """Returns custom UI field behaviour for Amazon Elastic MapReduce Connection."""
+ """Return custom UI field behaviour for Amazon Elastic MapReduce Connection."""
return {
"hidden_fields": ["host", "schema", "port", "login", "password"],
"relabeling": {
diff --git a/airflow/providers/amazon/aws/hooks/glue.py b/airflow/providers/amazon/aws/hooks/glue.py
index 0e8ccd6907..1d4c87d676 100644
--- a/airflow/providers/amazon/aws/hooks/glue.py
+++ b/airflow/providers/amazon/aws/hooks/glue.py
@@ -162,7 +162,7 @@ class GlueJobHook(AwsBaseHook):
run_kwargs: dict | None = None,
) -> dict[str, str]:
"""
- Initializes connection with AWS Glue to run job.
+ Initialize connection with AWS Glue to run job.
.. seealso::
- :external+boto3:py:meth:`Glue.Client.start_job_run`
@@ -196,7 +196,11 @@ class GlueJobHook(AwsBaseHook):
return job_run["JobRun"]["JobRunState"]
async def async_get_job_state(self, job_name: str, run_id: str) -> str:
- """The async version of get_job_state."""
+ """
+ Get state of the Glue job; the job state can be running, finished, failed, stopped or timeout.
+
+ The async version of get_job_state.
+ """
async with self.async_conn as client:
job_run = await client.get_job_run(JobName=job_name, RunId=run_id)
return job_run["JobRun"]["JobRunState"]
@@ -208,7 +212,7 @@ class GlueJobHook(AwsBaseHook):
continuation_tokens: LogContinuationTokens,
):
"""
- Prints the latest job logs to the Airflow task log and updates the continuation tokens.
+ Print the latest job logs to the Airflow task log and updates the continuation tokens.
:param continuation_tokens: the tokens where to resume from when reading logs.
The object gets updated with the new tokens by this method.
@@ -217,7 +221,7 @@ class GlueJobHook(AwsBaseHook):
paginator = log_client.get_paginator("filter_log_events")
def display_logs_from(log_group: str, continuation_token: str | None) -> str | None:
- """Internal method to mutualize iteration over the 2 different log streams glue jobs write to."""
+ """Mutualize iteration over the 2 different log streams glue jobs write to."""
fetched_logs = []
next_token = continuation_token
try:
@@ -305,7 +309,7 @@ class GlueJobHook(AwsBaseHook):
verbose: bool,
next_log_tokens: GlueJobHook.LogContinuationTokens,
) -> dict | None:
- """Helper function to process Glue Job state while polling. Used by both sync and async methods."""
+ """Process Glue Job state while polling; used by both sync and async methods."""
failed_states = ["FAILED", "TIMEOUT"]
finished_states = ["SUCCEEDED", "STOPPED"]
@@ -333,7 +337,7 @@ class GlueJobHook(AwsBaseHook):
def has_job(self, job_name) -> bool:
"""
- Checks if the job already exists.
+ Check if the job already exists.
.. seealso::
- :external+boto3:py:meth:`Glue.Client.get_job`
@@ -351,7 +355,7 @@ class GlueJobHook(AwsBaseHook):
def update_job(self, **job_kwargs) -> bool:
"""
- Updates job configurations.
+ Update job configurations.
.. seealso::
- :external+boto3:py:meth:`Glue.Client.update_job`
@@ -393,7 +397,7 @@ class GlueJobHook(AwsBaseHook):
def create_or_update_glue_job(self) -> str | None:
"""
- Creates (or updates) and returns the Job name.
+ Create (or update) and return the Job name.
.. seealso::
- :external+boto3:py:meth:`Glue.Client.update_job`
diff --git a/airflow/providers/amazon/aws/hooks/glue_catalog.py b/airflow/providers/amazon/aws/hooks/glue_catalog.py
index a5d434879e..0dac059d74 100644
--- a/airflow/providers/amazon/aws/hooks/glue_catalog.py
+++ b/airflow/providers/amazon/aws/hooks/glue_catalog.py
@@ -51,7 +51,7 @@ class GlueCatalogHook(AwsBaseHook):
max_items: int | None = None,
) -> set[tuple]:
"""
- Retrieves the partition values for a table.
+ Retrieve the partition values for a table.
.. seealso::
- :external+boto3:py:class:`Glue.Paginator.GetPartitions`
@@ -86,7 +86,7 @@ class GlueCatalogHook(AwsBaseHook):
def check_for_partition(self, database_name: str, table_name: str, expression: str) -> bool:
"""
- Checks whether a partition exists.
+ Check whether a partition exists.
.. code-block:: python
@@ -138,7 +138,7 @@ class GlueCatalogHook(AwsBaseHook):
def get_partition(self, database_name: str, table_name: str, partition_values: list[str]) -> dict:
"""
- Gets a Partition.
+ Get a Partition.
.. seealso::
- :external+boto3:py:meth:`Glue.Client.get_partition`
@@ -167,7 +167,7 @@ class GlueCatalogHook(AwsBaseHook):
def create_partition(self, database_name: str, table_name: str, partition_input: dict) -> dict:
"""
- Creates a new Partition.
+ Create a new Partition.
.. seealso::
- :external+boto3:py:meth:`Glue.Client.create_partition`
diff --git a/airflow/providers/amazon/aws/hooks/glue_crawler.py b/airflow/providers/amazon/aws/hooks/glue_crawler.py
index e0022c1e49..66836f9d58 100644
--- a/airflow/providers/amazon/aws/hooks/glue_crawler.py
+++ b/airflow/providers/amazon/aws/hooks/glue_crawler.py
@@ -49,7 +49,7 @@ class GlueCrawlerHook(AwsBaseHook):
def has_crawler(self, crawler_name) -> bool:
"""
- Checks if the crawler already exists.
+ Check if the crawler already exists.
:param crawler_name: unique crawler name per AWS account
:return: Returns True if the crawler already exists and False if not.
@@ -64,7 +64,7 @@ class GlueCrawlerHook(AwsBaseHook):
def get_crawler(self, crawler_name: str) -> dict:
"""
- Gets crawler configurations.
+ Get crawler configurations.
.. seealso::
- :external+boto3:py:meth:`Glue.Client.get_crawler`
@@ -76,7 +76,7 @@ class GlueCrawlerHook(AwsBaseHook):
def update_crawler(self, **crawler_kwargs) -> bool:
"""
- Updates crawler configurations.
+ Update crawler configurations.
.. seealso::
- :external+boto3:py:meth:`Glue.Client.update_crawler`
@@ -105,7 +105,7 @@ class GlueCrawlerHook(AwsBaseHook):
def update_tags(self, crawler_name: str, crawler_tags: dict) -> bool:
"""
- Updates crawler tags.
+ Update crawler tags.
.. seealso::
- :external+boto3:py:meth:`Glue.Client.tag_resource`
@@ -145,7 +145,7 @@ class GlueCrawlerHook(AwsBaseHook):
def create_crawler(self, **crawler_kwargs) -> str:
"""
- Creates an AWS Glue Crawler.
+ Create an AWS Glue Crawler.
.. seealso::
- :external+boto3:py:meth:`Glue.Client.create_crawler`
diff --git a/airflow/providers/amazon/aws/hooks/lambda_function.py b/airflow/providers/amazon/aws/hooks/lambda_function.py
index 58ecac8bcc..fcb24f7763 100644
--- a/airflow/providers/amazon/aws/hooks/lambda_function.py
+++ b/airflow/providers/amazon/aws/hooks/lambda_function.py
@@ -104,7 +104,7 @@ class LambdaHook(AwsBaseHook):
architectures: list[str] | None = None,
) -> dict:
"""
- Creates a Lambda function.
+ Create a Lambda function.
.. seealso::
- :external+boto3:py:meth:`Lambda.Client.create_function`
diff --git a/airflow/providers/amazon/aws/hooks/logs.py b/airflow/providers/amazon/aws/hooks/logs.py
index 8e494823bd..5e38ad32e3 100644
--- a/airflow/providers/amazon/aws/hooks/logs.py
+++ b/airflow/providers/amazon/aws/hooks/logs.py
@@ -67,7 +67,7 @@ class AwsLogsHook(AwsBaseHook):
end_time: int | None = None,
) -> Generator:
"""
- A generator for log items in a single stream; yields all items available at the current moment.
+ Return a generator for log items in a single stream; yields all items available at the current moment.
.. seealso::
- :external+boto3:py:meth:`CloudWatchLogs.Client.get_log_events`
diff --git a/airflow/providers/amazon/aws/hooks/quicksight.py b/airflow/providers/amazon/aws/hooks/quicksight.py
index 8cdf62db00..02ea582631 100644
--- a/airflow/providers/amazon/aws/hooks/quicksight.py
+++ b/airflow/providers/amazon/aws/hooks/quicksight.py
@@ -59,7 +59,7 @@ class QuickSightHook(AwsBaseHook):
check_interval: int = 30,
) -> dict:
"""
- Creates and starts a new SPICE ingestion for a dataset. Refreshes the SPICE datasets.
+ Create and start a new SPICE ingestion for a dataset; refresh the SPICE datasets.
.. seealso::
- :external+boto3:py:meth:`QuickSight.Client.create_ingestion`
@@ -121,7 +121,7 @@ class QuickSightHook(AwsBaseHook):
def get_error_info(self, aws_account_id: str, data_set_id: str, ingestion_id: str) -> dict | None:
"""
- Gets info about the error if any.
+ Get info about the error if any.
:param aws_account_id: An AWS Account ID
:param data_set_id: QuickSight Data Set ID
diff --git a/airflow/providers/amazon/aws/hooks/rds.py b/airflow/providers/amazon/aws/hooks/rds.py
index 0a1dd798da..ce2c0d313e 100644
--- a/airflow/providers/amazon/aws/hooks/rds.py
+++ b/airflow/providers/amazon/aws/hooks/rds.py
@@ -146,7 +146,7 @@ class RdsHook(AwsGenericHook["RDSClient"]):
def get_export_task_state(self, export_task_id: str) -> str:
"""
- Gets the current state of an RDS snapshot export to Amazon S3.
+ Get the current state of an RDS snapshot export to Amazon S3.
.. seealso::
- :external+boto3:py:meth:`RDS.Client.describe_export_tasks`
@@ -187,7 +187,7 @@ class RdsHook(AwsGenericHook["RDSClient"]):
def get_event_subscription_state(self, subscription_name: str) -> str:
"""
- Gets the current state of an RDS snapshot export to Amazon S3.
+ Get the current state of an RDS snapshot export to Amazon S3.
.. seealso::
- :external+boto3:py:meth:`RDS.Client.describe_event_subscriptions`
@@ -335,7 +335,7 @@ class RdsHook(AwsGenericHook["RDSClient"]):
max_attempts: int,
) -> None:
"""
- Polls the poke function for the current state until it reaches the target_state.
+ Poll the poke function for the current state until it reaches the target_state.
:param poke: A function that returns the current state of the target resource as a string.
:param target_state: Wait until this state is reached
diff --git a/airflow/providers/amazon/aws/hooks/redshift_data.py b/airflow/providers/amazon/aws/hooks/redshift_data.py
index ba7a61c28f..852b1e5c68 100644
--- a/airflow/providers/amazon/aws/hooks/redshift_data.py
+++ b/airflow/providers/amazon/aws/hooks/redshift_data.py
@@ -144,7 +144,7 @@ class RedshiftDataHook(AwsGenericHook["RedshiftDataAPIServiceClient"]):
poll_interval: int = 10,
) -> list[str] | None:
"""
- Helper method that returns the table primary key.
+ Return the table primary key.
Copied from ``RedshiftSQLHook.get_table_primary_key()``
diff --git a/airflow/providers/amazon/aws/hooks/redshift_sql.py b/airflow/providers/amazon/aws/hooks/redshift_sql.py
index ffba09b6fc..0b1c26fff5 100644
--- a/airflow/providers/amazon/aws/hooks/redshift_sql.py
+++ b/airflow/providers/amazon/aws/hooks/redshift_sql.py
@@ -64,7 +64,7 @@ class RedshiftSQLHook(DbApiHook):
@staticmethod
def get_ui_field_behaviour() -> dict:
- """Custom field behavior."""
+ """Get custom field behavior."""
return {
"hidden_fields": [],
"relabeling": {"login": "User", "schema": "Database"},
diff --git a/airflow/providers/amazon/aws/hooks/s3.py b/airflow/providers/amazon/aws/hooks/s3.py
index f0e98ed448..4e44a031b5 100644
--- a/airflow/providers/amazon/aws/hooks/s3.py
+++ b/airflow/providers/amazon/aws/hooks/s3.py
@@ -197,7 +197,7 @@ class S3Hook(AwsBaseHook):
@staticmethod
def parse_s3_url(s3url: str) -> tuple[str, str]:
"""
- Parses the S3 Url into a bucket name and key.
+ Parse the S3 Url into a bucket name and key.
See https://docs.aws.amazon.com/AmazonS3/latest/userguide/access-bucket-intro.html
for valid url formats.
@@ -300,7 +300,7 @@ class S3Hook(AwsBaseHook):
@provide_bucket_name
def get_bucket(self, bucket_name: str | None = None) -> S3Bucket:
"""
- Returns a :py:class:`S3.Bucket` object.
+ Return a :py:class:`S3.Bucket` object.
.. seealso::
- :external+boto3:py:meth:`S3.ServiceResource.Bucket`
@@ -319,7 +319,7 @@ class S3Hook(AwsBaseHook):
@provide_bucket_name
def create_bucket(self, bucket_name: str | None = None, region_name: str | None = None) -> None:
"""
- Creates an Amazon S3 bucket.
+ Create an Amazon S3 bucket.
.. seealso::
- :external+boto3:py:meth:`S3.Client.create_bucket`
@@ -345,7 +345,7 @@ class S3Hook(AwsBaseHook):
@provide_bucket_name
def check_for_prefix(self, prefix: str, delimiter: str, bucket_name: str | None = None) -> bool:
"""
- Checks that a prefix exists in a bucket.
+ Check that a prefix exists in a bucket.
:param bucket_name: the name of the bucket
:param prefix: a key prefix
@@ -368,7 +368,7 @@ class S3Hook(AwsBaseHook):
max_items: int | None = None,
) -> list:
"""
- Lists prefixes in a bucket under prefix.
+ List prefixes in a bucket under prefix.
.. seealso::
- :external+boto3:py:class:`S3.Paginator.ListObjectsV2`
@@ -405,7 +405,7 @@ class S3Hook(AwsBaseHook):
self, client: AioBaseClient, key: str, bucket_name: str | None = None
) -> dict[str, Any] | None:
"""
- Retrieves metadata of an object.
+ Retrieve metadata of an object.
:param client: aiobotocore client
:param bucket_name: Name of the bucket in which the file is stored
@@ -431,7 +431,7 @@ class S3Hook(AwsBaseHook):
max_items: int | None = None,
) -> list[Any]:
"""
- Lists prefixes in a bucket under prefix.
+ List prefixes in a bucket under prefix.
:param client: ClientCreatorContext
:param bucket_name: the name of the bucket
@@ -464,7 +464,7 @@ class S3Hook(AwsBaseHook):
@provide_bucket_name_async
async def get_file_metadata_async(self, client: AioBaseClient, bucket_name: str, key: str) -> list[Any]:
"""
- Gets a list of files that a key matching a wildcard expression exists in a bucket asynchronously.
+ Get a list of files that a key matching a wildcard expression exists in a bucket asynchronously.
:param client: aiobotocore client
:param bucket_name: the name of the bucket
@@ -520,7 +520,7 @@ class S3Hook(AwsBaseHook):
wildcard_match: bool,
) -> bool:
"""
- Checks for all keys in bucket and returns boolean value.
+ Check for all keys in bucket and returns boolean value.
:param client: aiobotocore client
:param bucket: the name of the bucket
@@ -539,7 +539,7 @@ class S3Hook(AwsBaseHook):
self, client: AioBaseClient, prefix: str, delimiter: str, bucket_name: str | None = None
) -> bool:
"""
- Checks that a prefix exists in a bucket.
+ Check that a prefix exists in a bucket.
:param bucket_name: the name of the bucket
:param prefix: a key prefix
@@ -567,7 +567,7 @@ class S3Hook(AwsBaseHook):
wildcard_match: bool,
delimiter: str | None = "/",
) -> list[Any]:
- """Gets a list of files in the bucket."""
+ """Get a list of files in the bucket."""
keys: list[Any] = []
for key in bucket_keys:
prefix = key
@@ -592,7 +592,7 @@ class S3Hook(AwsBaseHook):
max_items: int | None = None,
) -> list[str]:
"""
- Lists keys in a bucket under prefix and not containing delimiter.
+ List keys in a bucket under prefix and not containing delimiter.
:param bucket_name: the name of the bucket
:param prefix: a key prefix
@@ -758,7 +758,7 @@ class S3Hook(AwsBaseHook):
apply_wildcard: bool = False,
) -> list:
"""
- Lists keys in a bucket under prefix and not containing delimiter.
+ List keys in a bucket under prefix and not containing delimiter.
.. seealso::
- :external+boto3:py:class:`S3.Paginator.ListObjectsV2`
@@ -839,7 +839,7 @@ class S3Hook(AwsBaseHook):
max_items: int | None = None,
) -> list:
"""
- Lists metadata objects in a bucket under prefix.
+ List metadata objects in a bucket under prefix.
.. seealso::
- :external+boto3:py:class:`S3.Paginator.ListObjectsV2`
@@ -868,7 +868,7 @@ class S3Hook(AwsBaseHook):
@provide_bucket_name
def head_object(self, key: str, bucket_name: str | None = None) -> dict | None:
"""
- Retrieves metadata of an object.
+ Retrieve metadata of an object.
.. seealso::
- :external+boto3:py:meth:`S3.Client.head_object`
@@ -889,7 +889,7 @@ class S3Hook(AwsBaseHook):
@provide_bucket_name
def check_for_key(self, key: str, bucket_name: str | None = None) -> bool:
"""
- Checks if a key exists in a bucket.
+ Check if a key exists in a bucket.
.. seealso::
- :external+boto3:py:meth:`S3.Client.head_object`
@@ -905,7 +905,7 @@ class S3Hook(AwsBaseHook):
@provide_bucket_name
def get_key(self, key: str, bucket_name: str | None = None) -> S3ResourceObject:
"""
- Returns a :py:class:`S3.Object`.
+ Return a :py:class:`S3.Object`.
.. seealso::
- :external+boto3:py:meth:`S3.ServiceResource.Object`
@@ -928,7 +928,7 @@ class S3Hook(AwsBaseHook):
@provide_bucket_name
def read_key(self, key: str, bucket_name: str | None = None) -> str:
"""
- Reads a key from S3.
+ Read a key from S3.
.. seealso::
- :external+boto3:py:meth:`S3.Object.get`
@@ -952,7 +952,7 @@ class S3Hook(AwsBaseHook):
output_serialization: dict[str, Any] | None = None,
) -> str:
"""
- Reads a key with S3 Select.
+ Read a key with S3 Select.
.. seealso::
- :external+boto3:py:meth:`S3.Client.select_object_content`
@@ -992,7 +992,7 @@ class S3Hook(AwsBaseHook):
self, wildcard_key: str, bucket_name: str | None = None, delimiter: str = ""
) -> bool:
"""
- Checks that a key matching a wildcard expression exists in a bucket.
+ Check that a key matching a wildcard expression exists in a bucket.
:param wildcard_key: the path to the key
:param bucket_name: the name of the bucket
@@ -1010,7 +1010,7 @@ class S3Hook(AwsBaseHook):
self, wildcard_key: str, bucket_name: str | None = None, delimiter: str = ""
) -> S3ResourceObject | None:
"""
- Returns a boto3.s3.Object object matching the wildcard expression.
+ Return a boto3.s3.Object object matching the wildcard expression.
:param wildcard_key: the path to the key
:param bucket_name: the name of the bucket
@@ -1037,7 +1037,7 @@ class S3Hook(AwsBaseHook):
acl_policy: str | None = None,
) -> None:
"""
- Loads a local file to S3.
+ Load a local file to S3.
.. seealso::
- :external+boto3:py:meth:`S3.Client.upload_file`
@@ -1087,7 +1087,7 @@ class S3Hook(AwsBaseHook):
compression: str | None = None,
) -> None:
"""
- Loads a string to S3.
+ Load a string to S3.
This is provided as a convenience to drop a string in S3. It uses the
boto infrastructure to ship a file to s3.
@@ -1138,7 +1138,7 @@ class S3Hook(AwsBaseHook):
acl_policy: str | None = None,
) -> None:
"""
- Loads bytes to S3.
+ Load bytes to S3.
This is provided as a convenience to drop bytes data into S3. It uses the
boto infrastructure to ship a file to s3.
@@ -1172,7 +1172,7 @@ class S3Hook(AwsBaseHook):
acl_policy: str | None = None,
) -> None:
"""
- Loads a file object to S3.
+ Load a file object to S3.
.. seealso::
- :external+boto3:py:meth:`S3.Client.upload_fileobj`
@@ -1226,7 +1226,7 @@ class S3Hook(AwsBaseHook):
acl_policy: str | None = None,
) -> None:
"""
- Creates a copy of an object that is already stored in S3.
+ Create a copy of an object that is already stored in S3.
.. seealso::
- :external+boto3:py:meth:`S3.Client.copy_object`
@@ -1339,7 +1339,7 @@ class S3Hook(AwsBaseHook):
use_autogenerated_subdir: bool = True,
) -> str:
"""
- Downloads a file from the S3 location to the local file system.
+ Download a file from the S3 location to the local file system.
.. seealso::
- :external+boto3:py:meth:`S3.Object.download_fileobj`
@@ -1435,7 +1435,7 @@ class S3Hook(AwsBaseHook):
@provide_bucket_name
def get_bucket_tagging(self, bucket_name: str | None = None) -> list[dict[str, str]] | None:
"""
- Gets a List of tags from a bucket.
+ Get a List of tags from a bucket.
.. seealso::
- :external+boto3:py:meth:`S3.Client.get_bucket_tagging`
@@ -1461,7 +1461,7 @@ class S3Hook(AwsBaseHook):
bucket_name: str | None = None,
) -> None:
"""
- Overwrites the existing TagSet with provided tags; must provide a TagSet, a key/value pair, or both.
+ Overwrite the existing TagSet with provided tags; must provide a TagSet, a key/value pair, or both.
.. seealso::
- :external+boto3:py:meth:`S3.Client.put_bucket_tagging`
@@ -1498,7 +1498,7 @@ class S3Hook(AwsBaseHook):
@provide_bucket_name
def delete_bucket_tagging(self, bucket_name: str | None = None) -> None:
"""
- Deletes all tags from a bucket.
+ Delete all tags from a bucket.
.. seealso::
- :external+boto3:py:meth:`S3.Client.delete_bucket_tagging`
diff --git a/airflow/providers/amazon/aws/hooks/step_function.py b/airflow/providers/amazon/aws/hooks/step_function.py
index 21bc51d8dc..2d760ee891 100644
--- a/airflow/providers/amazon/aws/hooks/step_function.py
+++ b/airflow/providers/amazon/aws/hooks/step_function.py
@@ -71,7 +71,7 @@ class StepFunctionHook(AwsBaseHook):
def describe_execution(self, execution_arn: str) -> dict:
"""
- Describes a State Machine Execution.
+ Describe a State Machine Execution.
.. seealso::
- :external+boto3:py:meth:`SFN.Client.describe_execution`
diff --git a/airflow/providers/amazon/aws/links/emr.py b/airflow/providers/amazon/aws/links/emr.py
index f5def55f87..cacb4ff5fb 100644
--- a/airflow/providers/amazon/aws/links/emr.py
+++ b/airflow/providers/amazon/aws/links/emr.py
@@ -51,7 +51,7 @@ def get_log_uri(
*, cluster: dict[str, Any] | None = None, emr_client: boto3.client = None, job_flow_id: str | None = None
) -> str | None:
"""
- Retrieves the S3 URI to the EMR Job logs.
+ Retrieve the S3 URI to the EMR Job logs.
Requires either the output of a describe_cluster call or both an EMR Client and a job_flow_id..
"""
diff --git a/airflow/providers/amazon/aws/log/s3_task_handler.py b/airflow/providers/amazon/aws/log/s3_task_handler.py
index 4a813715fd..ea7266c28c 100644
--- a/airflow/providers/amazon/aws/log/s3_task_handler.py
+++ b/airflow/providers/amazon/aws/log/s3_task_handler.py
@@ -170,7 +170,7 @@ class S3TaskHandler(FileTaskHandler, LoggingMixin):
def s3_read(self, remote_log_location: str, return_error: bool = False) -> str:
"""
- Returns the log found at the remote_log_location. Return '' if no logs are found or there is an error.
+ Return the log found at the remote_log_location or '' if no logs are found or there is an error.
:param remote_log_location: the log's location in remote storage
:param return_error: if True, returns a string error message if an
diff --git a/airflow/providers/amazon/aws/operators/datasync.py b/airflow/providers/amazon/aws/operators/datasync.py
index 76b288a548..8f3c0ddf72 100644
--- a/airflow/providers/amazon/aws/operators/datasync.py
+++ b/airflow/providers/amazon/aws/operators/datasync.py
@@ -388,7 +388,7 @@ class DataSyncOperator(BaseOperator):
self._cancel_datasync_task_execution()
def _delete_datasync_task(self) -> None:
- """Deletes an AWS DataSync Task."""
+ """Delete an AWS DataSync Task."""
if not self.task_arn:
return
diff --git a/airflow/providers/amazon/aws/operators/dms.py b/airflow/providers/amazon/aws/operators/dms.py
index afa81ab040..8a32f33f18 100644
--- a/airflow/providers/amazon/aws/operators/dms.py
+++ b/airflow/providers/amazon/aws/operators/dms.py
@@ -88,7 +88,7 @@ class DmsCreateTaskOperator(BaseOperator):
def execute(self, context: Context):
"""
- Creates AWS DMS replication task from Airflow.
+ Create AWS DMS replication task from Airflow.
:return: replication task arn
"""
@@ -141,7 +141,7 @@ class DmsDeleteTaskOperator(BaseOperator):
def execute(self, context: Context):
"""
- Deletes AWS DMS replication task from Airflow.
+ Delete AWS DMS replication task from Airflow.
:return: replication task arn
"""
@@ -183,7 +183,7 @@ class DmsDescribeTasksOperator(BaseOperator):
def execute(self, context: Context) -> tuple[str | None, list]:
"""
- Describes AWS DMS replication tasks from Airflow.
+ Describe AWS DMS replication tasks from Airflow.
:return: Marker and list of replication tasks
"""
@@ -235,7 +235,7 @@ class DmsStartTaskOperator(BaseOperator):
def execute(self, context: Context):
"""
- Starts AWS DMS replication task from Airflow.
+ Start AWS DMS replication task from Airflow.
:return: replication task arn
"""
@@ -282,7 +282,7 @@ class DmsStopTaskOperator(BaseOperator):
def execute(self, context: Context):
"""
- Stops AWS DMS replication task from Airflow.
+ Stop AWS DMS replication task from Airflow.
:return: replication task arn
"""
diff --git a/airflow/providers/amazon/aws/operators/eks.py b/airflow/providers/amazon/aws/operators/eks.py
index e7931cc07a..557bf29f5f 100644
--- a/airflow/providers/amazon/aws/operators/eks.py
+++ b/airflow/providers/amazon/aws/operators/eks.py
@@ -735,7 +735,7 @@ class EksDeleteClusterOperator(BaseOperator):
def delete_any_nodegroups(self, eks_hook) -> None:
"""
- Deletes all Amazon EKS managed node groups for a provided Amazon EKS Cluster.
+ Delete all Amazon EKS managed node groups for a provided Amazon EKS Cluster.
Amazon EKS managed node groups can be deleted in parallel, so we can send all
delete commands in bulk and move on once the count of nodegroups is zero.
@@ -752,7 +752,7 @@ class EksDeleteClusterOperator(BaseOperator):
def delete_any_fargate_profiles(self, eks_hook) -> None:
"""
- Deletes all EKS Fargate profiles for a provided Amazon EKS Cluster.
+ Delete all EKS Fargate profiles for a provided Amazon EKS Cluster.
EKS Fargate profiles must be deleted one at a time, so we must wait
for one to be deleted before sending the next delete command.
diff --git a/airflow/providers/amazon/aws/operators/glue_crawler.py b/airflow/providers/amazon/aws/operators/glue_crawler.py
index c9fb298ee7..99a705b6ea 100644
--- a/airflow/providers/amazon/aws/operators/glue_crawler.py
+++ b/airflow/providers/amazon/aws/operators/glue_crawler.py
@@ -80,7 +80,7 @@ class GlueCrawlerOperator(BaseOperator):
def execute(self, context: Context):
"""
- Executes AWS Glue Crawler from Airflow.
+ Execute AWS Glue Crawler from Airflow.
:return: the name of the current glue crawler.
"""
diff --git a/airflow/providers/amazon/aws/operators/lambda_function.py b/airflow/providers/amazon/aws/operators/lambda_function.py
index 28b6313204..e472deaaac 100644
--- a/airflow/providers/amazon/aws/operators/lambda_function.py
+++ b/airflow/providers/amazon/aws/operators/lambda_function.py
@@ -169,7 +169,7 @@ class LambdaInvokeFunctionOperator(BaseOperator):
def execute(self, context: Context):
"""
- Invokes the target AWS Lambda function from Airflow.
+ Invoke the target AWS Lambda function from Airflow.
:return: The response payload from the function, or an error object.
"""
diff --git a/airflow/providers/amazon/aws/operators/sagemaker.py b/airflow/providers/amazon/aws/operators/sagemaker.py
index 7e1a0a3567..376c1051fb 100644
--- a/airflow/providers/amazon/aws/operators/sagemaker.py
+++ b/airflow/providers/amazon/aws/operators/sagemaker.py
@@ -96,7 +96,7 @@ class SageMakerBaseOperator(BaseOperator):
self.parse_integer(self.config, field)
def expand_role(self) -> None:
- """Placeholder for calling boto3's `expand_role`, which expands an IAM role name into an ARN."""
+ """Call boto3's `expand_role`, which expands an IAM role name into an ARN."""
def preprocess_config(self) -> None:
"""Process the config into a usable form."""
@@ -122,7 +122,7 @@ class SageMakerBaseOperator(BaseOperator):
self, proposed_name: str, fail_if_exists: bool, describe_func: Callable[[str], Any]
) -> str:
"""
- Returns the proposed name if it doesn't already exist, otherwise returns it with a timestamp suffix.
+ Return the proposed name if it doesn't already exist, otherwise returns it with a timestamp suffix.
:param proposed_name: Base name.
:param fail_if_exists: Will throw an error if a job with that name already exists
@@ -142,7 +142,7 @@ class SageMakerBaseOperator(BaseOperator):
return job_name
def _check_if_job_exists(self, job_name, describe_func: Callable[[str], Any]) -> bool:
- """Returns True if job exists, False otherwise."""
+ """Return True if job exists, False otherwise."""
try:
describe_func(job_name)
self.log.info("Found existing job with name '%s'.", job_name)
@@ -248,7 +248,7 @@ class SageMakerProcessingOperator(SageMakerBaseOperator):
self.integer_fields.append(["StoppingCondition", "MaxRuntimeInSeconds"])
def expand_role(self) -> None:
- """Expands an IAM role name into an ARN."""
+ """Expand an IAM role name into an ARN."""
if "RoleArn" in self.config:
hook = AwsBaseHook(self.aws_conn_id, client_type="iam")
self.config["RoleArn"] = hook.expand_role(self.config["RoleArn"])
@@ -306,7 +306,7 @@ class SageMakerProcessingOperator(SageMakerBaseOperator):
return {"Processing": self.serialized_job}
def get_openlineage_facets_on_complete(self, task_instance) -> OperatorLineage:
- """Returns OpenLineage data gathered from SageMaker's API response saved by processing job."""
+ """Return OpenLineage data gathered from SageMaker's API response saved by processing job."""
from airflow.providers.openlineage.extractors.base import OperatorLineage
inputs = []
@@ -462,7 +462,7 @@ class SageMakerEndpointOperator(SageMakerBaseOperator):
]
def expand_role(self) -> None:
- """Expands an IAM role name into an ARN."""
+ """Expand an IAM role name into an ARN."""
if "Model" not in self.config:
return
hook = AwsBaseHook(self.aws_conn_id, client_type="iam")
@@ -640,7 +640,7 @@ class SageMakerTransformOperator(SageMakerBaseOperator):
field.pop(0)
def expand_role(self) -> None:
- """Expands an IAM role name into an ARN."""
+ """Expand an IAM role name into an ARN."""
if "Model" not in self.config:
return
config = self.config["Model"]
@@ -717,7 +717,7 @@ class SageMakerTransformOperator(SageMakerBaseOperator):
return {"Model": self.serialized_model, "Transform": self.serialized_tranform}
def get_openlineage_facets_on_complete(self, task_instance) -> OperatorLineage:
- """Returns OpenLineage data gathered from SageMaker's API response saved by transform job."""
+ """Return OpenLineage data gathered from SageMaker's API response saved by transform job."""
from airflow.providers.openlineage.extractors import OperatorLineage
model_package_arn = None
@@ -815,7 +815,7 @@ class SageMakerTuningOperator(SageMakerBaseOperator):
self.deferrable = deferrable
def expand_role(self) -> None:
- """Expands an IAM role name into an ARN."""
+ """Expand an IAM role name into an ARN."""
if "TrainingJobDefinition" in self.config:
config = self.config["TrainingJobDefinition"]
if "RoleArn" in config:
@@ -905,7 +905,7 @@ class SageMakerModelOperator(SageMakerBaseOperator):
super().__init__(config=config, aws_conn_id=aws_conn_id, **kwargs)
def expand_role(self) -> None:
- """Expands an IAM role name into an ARN."""
+ """Expand an IAM role name into an ARN."""
if "ExecutionRoleArn" in self.config:
hook = AwsBaseHook(self.aws_conn_id, client_type="iam")
self.config["ExecutionRoleArn"] = hook.expand_role(self.config["ExecutionRoleArn"])
@@ -995,7 +995,7 @@ class SageMakerTrainingOperator(SageMakerBaseOperator):
self.serialized_training_data: dict
def expand_role(self) -> None:
- """Expands an IAM role name into an ARN."""
+ """Expand an IAM role name into an ARN."""
if "RoleArn" in self.config:
hook = AwsBaseHook(self.aws_conn_id, client_type="iam")
self.config["RoleArn"] = hook.expand_role(self.config["RoleArn"])
@@ -1069,7 +1069,7 @@ class SageMakerTrainingOperator(SageMakerBaseOperator):
return {"Training": self.serialized_training_data}
def get_openlineage_facets_on_complete(self, task_instance) -> OperatorLineage:
- """Returns OpenLineage data gathered from SageMaker's API response saved by training job."""
+ """Return OpenLineage data gathered from SageMaker's API response saved by training job."""
from airflow.providers.openlineage.extractors import OperatorLineage
inputs = []
diff --git a/airflow/providers/amazon/aws/sensors/batch.py b/airflow/providers/amazon/aws/sensors/batch.py
index 454d123f94..e493343c73 100644
--- a/airflow/providers/amazon/aws/sensors/batch.py
+++ b/airflow/providers/amazon/aws/sensors/batch.py
@@ -110,7 +110,7 @@ class BatchSensor(BaseSensorOperator):
def execute_complete(self, context: Context, event: dict[str, Any]) -> None:
"""
- Callback for when the trigger fires - returns immediately.
+ Execute when the trigger fires - returns immediately.
Relies on trigger to throw an exception, otherwise it assumes execution was successful.
"""
diff --git a/airflow/providers/amazon/aws/sensors/glue_catalog_partition.py b/airflow/providers/amazon/aws/sensors/glue_catalog_partition.py
index 2f5a002084..20f29f5251 100644
--- a/airflow/providers/amazon/aws/sensors/glue_catalog_partition.py
+++ b/airflow/providers/amazon/aws/sensors/glue_catalog_partition.py
@@ -76,7 +76,7 @@ class GlueCatalogPartitionSensor(BaseSensorOperator):
self.database_name = database_name
def poke(self, context: Context):
- """Checks for existence of the partition in the AWS Glue Catalog table."""
+ """Check for existence of the partition in the AWS Glue Catalog table."""
if "." in self.table_name:
self.database_name, self.table_name = self.table_name.split(".")
self.log.info(
@@ -87,7 +87,7 @@ class GlueCatalogPartitionSensor(BaseSensorOperator):
@deprecated(reason="use `hook` property instead.")
def get_hook(self) -> GlueCatalogHook:
- """Gets the GlueCatalogHook."""
+ """Get the GlueCatalogHook."""
return self.hook
@cached_property
diff --git a/airflow/providers/amazon/aws/sensors/glue_crawler.py b/airflow/providers/amazon/aws/sensors/glue_crawler.py
index 4b3320b0f4..5656569577 100644
--- a/airflow/providers/amazon/aws/sensors/glue_crawler.py
+++ b/airflow/providers/amazon/aws/sensors/glue_crawler.py
@@ -69,7 +69,7 @@ class GlueCrawlerSensor(BaseSensorOperator):
@deprecated(reason="use `hook` property instead.")
def get_hook(self) -> GlueCrawlerHook:
- """Returns a new or pre-existing GlueCrawlerHook."""
+ """Return a new or pre-existing GlueCrawlerHook."""
return self.hook
@cached_property
diff --git a/airflow/providers/amazon/aws/sensors/s3.py b/airflow/providers/amazon/aws/sensors/s3.py
index c245ab3744..37f18c7621 100644
--- a/airflow/providers/amazon/aws/sensors/s3.py
+++ b/airflow/providers/amazon/aws/sensors/s3.py
@@ -164,7 +164,7 @@ class S3KeySensor(BaseSensorOperator):
def execute_complete(self, context: Context, event: dict[str, Any]) -> bool | None:
"""
- Callback for when the trigger fires - returns immediately.
+ Execute when the trigger fires - returns immediately.
Relies on trigger to throw an exception, otherwise it assumes execution was successful.
"""
@@ -355,7 +355,7 @@ class S3KeysUnchangedSensor(BaseSensorOperator):
def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> None:
"""
- Callback for when the trigger fires - returns immediately.
+ Execute when the trigger fires - returns immediately.
Relies on trigger to throw an exception, otherwise it assumes execution was successful.
"""
diff --git a/airflow/providers/amazon/aws/sensors/sagemaker.py b/airflow/providers/amazon/aws/sensors/sagemaker.py
index 6ede44bc43..b960d1cd03 100644
--- a/airflow/providers/amazon/aws/sensors/sagemaker.py
+++ b/airflow/providers/amazon/aws/sensors/sagemaker.py
@@ -71,23 +71,23 @@ class SageMakerBaseSensor(BaseSensorOperator):
return True
def non_terminal_states(self) -> set[str]:
- """Placeholder for returning states with should not terminate."""
+ """Return states with should not terminate."""
raise NotImplementedError("Please implement non_terminal_states() in subclass")
def failed_states(self) -> set[str]:
- """Placeholder for returning states with are considered failed."""
+ """Return states with are considered failed."""
raise NotImplementedError("Please implement failed_states() in subclass")
def get_sagemaker_response(self) -> dict:
- """Placeholder for checking status of a SageMaker task."""
+ """Check status of a SageMaker task."""
raise NotImplementedError("Please implement get_sagemaker_response() in subclass")
def get_failed_reason_from_response(self, response: dict) -> str:
- """Placeholder for extracting the reason for failure from an AWS response."""
+ """Extract the reason for failure from an AWS response."""
return "Unknown"
def state_from_response(self, response: dict) -> str:
- """Placeholder for extracting the state from an AWS response."""
+ """Extract the state from an AWS response."""
raise NotImplementedError("Please implement state_from_response() in subclass")
diff --git a/airflow/providers/amazon/aws/transfers/imap_attachment_to_s3.py b/airflow/providers/amazon/aws/transfers/imap_attachment_to_s3.py
index a8491f3805..f23421502c 100644
--- a/airflow/providers/amazon/aws/transfers/imap_attachment_to_s3.py
+++ b/airflow/providers/amazon/aws/transfers/imap_attachment_to_s3.py
@@ -77,7 +77,7 @@ class ImapAttachmentToS3Operator(BaseOperator):
def execute(self, context: Context) -> None:
"""
- This function executes the transfer from the email server (via imap) into s3.
+ Execute the transfer from the email server (via imap) into s3.
:param context: The context while executing.
"""
diff --git a/airflow/providers/amazon/aws/transfers/s3_to_sftp.py b/airflow/providers/amazon/aws/transfers/s3_to_sftp.py
index c6499b77e6..596edb6016 100644
--- a/airflow/providers/amazon/aws/transfers/s3_to_sftp.py
+++ b/airflow/providers/amazon/aws/transfers/s3_to_sftp.py
@@ -69,7 +69,7 @@ class S3ToSFTPOperator(BaseOperator):
@staticmethod
def get_s3_key(s3_key: str) -> str:
- """This parses the correct format for S3 keys regardless of how the S3 url is passed."""
+ """Parse the correct format for S3 keys regardless of how the S3 url is passed."""
parsed_s3_key = urlsplit(s3_key)
return parsed_s3_key.path.lstrip("/")
diff --git a/airflow/providers/amazon/aws/transfers/sftp_to_s3.py b/airflow/providers/amazon/aws/transfers/sftp_to_s3.py
index 2d040fd04a..322110dadb 100644
--- a/airflow/providers/amazon/aws/transfers/sftp_to_s3.py
+++ b/airflow/providers/amazon/aws/transfers/sftp_to_s3.py
@@ -74,7 +74,7 @@ class SFTPToS3Operator(BaseOperator):
@staticmethod
def get_s3_key(s3_key: str) -> str:
- """This parses the correct format for S3 keys regardless of how the S3 url is passed."""
+ """Parse the correct format for S3 keys regardless of how the S3 url is passed."""
parsed_s3_key = urlsplit(s3_key)
return parsed_s3_key.path.lstrip("/")
diff --git a/airflow/providers/amazon/aws/triggers/batch.py b/airflow/providers/amazon/aws/triggers/batch.py
index 40b230221f..774ce3c4bf 100644
--- a/airflow/providers/amazon/aws/triggers/batch.py
+++ b/airflow/providers/amazon/aws/triggers/batch.py
@@ -58,7 +58,7 @@ class BatchOperatorTrigger(BaseTrigger):
self.poll_interval = poll_interval
def serialize(self) -> tuple[str, dict[str, Any]]:
- """Serializes BatchOperatorTrigger arguments and classpath."""
+ """Serialize BatchOperatorTrigger arguments and classpath."""
return (
"airflow.providers.amazon.aws.triggers.batch.BatchOperatorTrigger",
{
@@ -139,7 +139,7 @@ class BatchSensorTrigger(BaseTrigger):
self.poke_interval = poke_interval
def serialize(self) -> tuple[str, dict[str, Any]]:
- """Serializes BatchSensorTrigger arguments and classpath."""
+ """Serialize BatchSensorTrigger arguments and classpath."""
return (
"airflow.providers.amazon.aws.triggers.batch.BatchSensorTrigger",
{
diff --git a/airflow/providers/amazon/aws/triggers/ecs.py b/airflow/providers/amazon/aws/triggers/ecs.py
index 4a380d23f0..ea183e183d 100644
--- a/airflow/providers/amazon/aws/triggers/ecs.py
+++ b/airflow/providers/amazon/aws/triggers/ecs.py
@@ -186,7 +186,7 @@ class TaskDoneTrigger(BaseTrigger):
async def _forward_logs(self, logs_client, next_token: str | None = None) -> str | None:
"""
- Reads logs from the cloudwatch stream and prints them to the task logs.
+ Read logs from the cloudwatch stream and print them to the task logs.
:return: the token to pass to the next iteration to resume where we started.
"""
diff --git a/airflow/providers/amazon/aws/triggers/eks.py b/airflow/providers/amazon/aws/triggers/eks.py
index b5419b76e7..01f66d4567 100644
--- a/airflow/providers/amazon/aws/triggers/eks.py
+++ b/airflow/providers/amazon/aws/triggers/eks.py
@@ -138,7 +138,7 @@ class EksDeleteClusterTrigger(AwsBaseWaiterTrigger):
async def delete_any_nodegroups(self, client) -> None:
"""
- Deletes all EKS Nodegroups for a provided Amazon EKS Cluster.
+ Delete all EKS Nodegroups for a provided Amazon EKS Cluster.
All the EKS Nodegroups are deleted simultaneously. We wait for
all Nodegroups to be deleted before returning.
@@ -167,7 +167,7 @@ class EksDeleteClusterTrigger(AwsBaseWaiterTrigger):
async def delete_any_fargate_profiles(self, client) -> None:
"""
- Deletes all EKS Fargate profiles for a provided Amazon EKS Cluster.
+ Delete all EKS Fargate profiles for a provided Amazon EKS Cluster.
EKS Fargate profiles must be deleted one at a time, so we must wait
for one to be deleted before sending the next delete command.
diff --git a/airflow/providers/amazon/aws/triggers/sagemaker.py b/airflow/providers/amazon/aws/triggers/sagemaker.py
index 7165276c4c..ae24f78fc5 100644
--- a/airflow/providers/amazon/aws/triggers/sagemaker.py
+++ b/airflow/providers/amazon/aws/triggers/sagemaker.py
@@ -59,7 +59,7 @@ class SageMakerTrigger(BaseTrigger):
self.aws_conn_id = aws_conn_id
def serialize(self) -> tuple[str, dict[str, Any]]:
- """Serializes SagemakerTrigger arguments and classpath."""
+ """Serialize SagemakerTrigger arguments and classpath."""
return (
"airflow.providers.amazon.aws.triggers.sagemaker.SageMakerTrigger",
{
diff --git a/airflow/providers/amazon/aws/utils/task_log_fetcher.py b/airflow/providers/amazon/aws/utils/task_log_fetcher.py
index f184df6234..e5cac2f426 100644
--- a/airflow/providers/amazon/aws/utils/task_log_fetcher.py
+++ b/airflow/providers/amazon/aws/utils/task_log_fetcher.py
@@ -92,7 +92,7 @@ class AwsTaskLogFetcher(Thread):
def get_last_log_messages(self, number_messages) -> list:
"""
- Gets the last logs messages in one single request.
+ Get the last logs messages in one single request.
NOTE: some restrictions apply:
- if logs are too old, the response will be empty
diff --git a/airflow/providers/apache/beam/hooks/beam.py b/airflow/providers/apache/beam/hooks/beam.py
index 1ddb6d6fdc..762dd2f07b 100644
--- a/airflow/providers/apache/beam/hooks/beam.py
+++ b/airflow/providers/apache/beam/hooks/beam.py
@@ -60,7 +60,7 @@ class BeamRunnerType:
def beam_options_to_args(options: dict) -> list[str]:
"""
- Returns a formatted pipeline options from a dictionary of arguments.
+ Return a formatted pipeline options from a dictionary of arguments.
The logic of this method should be compatible with Apache Beam:
https://github.com/apache/beam/blob/b56740f0e8cd80c2873412847d0b336837429fb9/sdks/python/
@@ -90,7 +90,7 @@ def process_fd(
process_line_callback: Callable[[str], None] | None = None,
):
"""
- Prints output to logs.
+ Print output to logs.
:param proc: subprocess.
:param fd: File descriptor.
@@ -120,7 +120,7 @@ def run_beam_command(
working_directory: str | None = None,
) -> None:
"""
- Function responsible for running pipeline command in subprocess.
+ Run pipeline command in subprocess.
:param cmd: Parts of the command to be run in subprocess
:param process_line_callback: Optional callback which can be used to process
@@ -211,7 +211,7 @@ class BeamHook(BaseHook):
process_line_callback: Callable[[str], None] | None = None,
):
"""
- Starts Apache Beam python pipeline.
+ Start Apache Beam python pipeline.
:param variables: Variables passed to the pipeline.
:param py_file: Path to the python file to execute.
@@ -291,7 +291,7 @@ class BeamHook(BaseHook):
process_line_callback: Callable[[str], None] | None = None,
) -> None:
"""
- Starts Apache Beam Java pipeline.
+ Start Apache Beam Java pipeline.
:param variables: Variables passed to the job.
:param jar: Name of the jar for the pipeline
@@ -317,7 +317,7 @@ class BeamHook(BaseHook):
should_init_module: bool = False,
) -> None:
"""
- Starts Apache Beam Go pipeline with a source file.
+ Start Apache Beam Go pipeline with a source file.
:param variables: Variables passed to the job.
:param go_file: Path to the Go file with your beam pipeline.
@@ -361,7 +361,7 @@ class BeamHook(BaseHook):
process_line_callback: Callable[[str], None] | None = None,
) -> None:
"""
- Starts Apache Beam Go pipeline with an executable binary.
+ Start Apache Beam Go pipeline with an executable binary.
:param variables: Variables passed to the job.
:param launcher_binary: Path to the binary compiled for the launching platform.
@@ -401,7 +401,7 @@ class BeamAsyncHook(BeamHook):
@staticmethod
async def _create_tmp_dir(prefix: str) -> str:
- """Helper method to create temporary directory."""
+ """Create temporary directory."""
# Creating separate thread to create temporary directory
loop = asyncio.get_running_loop()
partial_func = functools.partial(tempfile.mkdtemp, prefix=prefix)
@@ -411,7 +411,7 @@ class BeamAsyncHook(BeamHook):
@staticmethod
async def _cleanup_tmp_dir(tmp_dir: str) -> None:
"""
- Helper method to delete temporary directory after finishing work with it.
+ Delete temporary directory after finishing work with it.
Is uses `rmtree` method to recursively remove the temporary directory.
"""
@@ -427,7 +427,7 @@ class BeamAsyncHook(BeamHook):
py_system_site_packages: bool = False,
):
"""
- Starts Apache Beam python pipeline.
+ Start Apache Beam python pipeline.
:param variables: Variables passed to the pipeline.
:param py_file: Path to the python file to execute.
@@ -524,7 +524,7 @@ class BeamAsyncHook(BeamHook):
working_directory: str | None = None,
) -> int:
"""
- Function responsible for running pipeline command in subprocess.
+ Run pipeline command in subprocess.
:param cmd: Parts of the command to be run in subprocess
:param working_directory: Working directory
diff --git a/airflow/providers/apache/beam/operators/beam.py b/airflow/providers/apache/beam/operators/beam.py
index 4cec4ba31a..583dbe12a7 100644
--- a/airflow/providers/apache/beam/operators/beam.py
+++ b/airflow/providers/apache/beam/operators/beam.py
@@ -401,7 +401,7 @@ class BeamRunPythonPipelineOperator(BeamBasePipelineOperator):
def execute_complete(self, context: Context, event: dict[str, Any]):
"""
- Callback for when the trigger fires - returns immediately.
+ Execute when the trigger fires - returns immediately.
Relies on trigger to throw an exception, otherwise it assumes execution was
successful.
diff --git a/airflow/providers/apache/beam/triggers/beam.py b/airflow/providers/apache/beam/triggers/beam.py
index 2926a17b45..9c29a9fbe2 100644
--- a/airflow/providers/apache/beam/triggers/beam.py
+++ b/airflow/providers/apache/beam/triggers/beam.py
@@ -68,7 +68,7 @@ class BeamPipelineTrigger(BaseTrigger):
self.runner = runner
def serialize(self) -> tuple[str, dict[str, Any]]:
- """Serializes BeamPipelineTrigger arguments and classpath."""
+ """Serialize BeamPipelineTrigger arguments and classpath."""
return (
"airflow.providers.apache.beam.triggers.beam.BeamPipelineTrigger",
{
@@ -83,7 +83,7 @@ class BeamPipelineTrigger(BaseTrigger):
)
async def run(self) -> AsyncIterator[TriggerEvent]: # type: ignore[override]
- """Gets current pipeline status and yields a TriggerEvent."""
+ """Get current pipeline status and yields a TriggerEvent."""
hook = self._get_async_hook()
while True:
try:
diff --git a/airflow/providers/apache/cassandra/hooks/cassandra.py b/airflow/providers/apache/cassandra/hooks/cassandra.py
index 84f10c85b2..999782e0da 100644
--- a/airflow/providers/apache/cassandra/hooks/cassandra.py
+++ b/airflow/providers/apache/cassandra/hooks/cassandra.py
@@ -125,25 +125,25 @@ class CassandraHook(BaseHook, LoggingMixin):
self.session = None
def get_conn(self) -> Session:
- """Returns a cassandra Session object."""
+ """Return a cassandra Session object."""
if self.session and not self.session.is_shutdown:
return self.session
self.session = self.cluster.connect(self.keyspace)
return self.session
def get_cluster(self) -> Cluster:
- """Returns Cassandra cluster."""
+ """Return Cassandra cluster."""
return self.cluster
def shutdown_cluster(self) -> None:
- """Closes all sessions and connections associated with this Cluster."""
+ """Close all sessions and connections associated with this Cluster."""
if not self.cluster.is_shutdown:
self.cluster.shutdown()
@staticmethod
def get_lb_policy(policy_name: str, policy_args: dict[str, Any]) -> Policy:
"""
- Creates load balancing policy.
+ Create load balancing policy.
:param policy_name: Name of the policy to use.
:param policy_args: Parameters for the policy.
@@ -177,7 +177,7 @@ class CassandraHook(BaseHook, LoggingMixin):
def table_exists(self, table: str) -> bool:
"""
- Checks if a table exists in Cassandra.
+ Check if a table exists in Cassandra.
:param table: Target Cassandra table.
Use dot notation to target a specific keyspace.
@@ -190,7 +190,7 @@ class CassandraHook(BaseHook, LoggingMixin):
def record_exists(self, table: str, keys: dict[str, str]) -> bool:
"""
- Checks if a record exists in Cassandra.
+ Check if a record exists in Cassandra.
:param table: Target Cassandra table.
Use dot notation to target a specific keyspace.
diff --git a/airflow/providers/apache/drill/hooks/drill.py b/airflow/providers/apache/drill/hooks/drill.py
index 2f2bfa3273..962bd93aea 100644
--- a/airflow/providers/apache/drill/hooks/drill.py
+++ b/airflow/providers/apache/drill/hooks/drill.py
@@ -65,7 +65,7 @@ class DrillHook(DbApiHook):
def get_uri(self) -> str:
"""
- Returns the connection URI.
+ Return the connection URI.
e.g: ``drill://localhost:8047/dfs``
"""
diff --git a/airflow/providers/apache/druid/transfers/hive_to_druid.py b/airflow/providers/apache/druid/transfers/hive_to_druid.py
index cf2f3d551c..2616ea4f5e 100644
--- a/airflow/providers/apache/druid/transfers/hive_to_druid.py
+++ b/airflow/providers/apache/druid/transfers/hive_to_druid.py
@@ -154,7 +154,7 @@ class HiveToDruidOperator(BaseOperator):
def construct_ingest_query(self, static_path: str, columns: list[str]) -> dict[str, Any]:
"""
- Builds an ingest query for an HDFS TSV load.
+ Build an ingest query for an HDFS TSV load.
:param static_path: The path on hdfs where the data is
:param columns: List of all the columns that are available
diff --git a/airflow/providers/apache/hdfs/hooks/webhdfs.py b/airflow/providers/apache/hdfs/hooks/webhdfs.py
index 319a1bd60b..e99e20127e 100644
--- a/airflow/providers/apache/hdfs/hooks/webhdfs.py
+++ b/airflow/providers/apache/hdfs/hooks/webhdfs.py
@@ -59,7 +59,7 @@ class WebHDFSHook(BaseHook):
def get_conn(self) -> Any:
"""
- Establishes a connection depending on the security mode set via config or environment variable.
+ Establish a connection depending on the security mode set via config or environment variable.
:return: a hdfscli InsecureClient or KerberosClient object.
"""
@@ -134,11 +134,11 @@ class WebHDFSHook(BaseHook):
def load_file(
self, source: str, destination: str, overwrite: bool = True, parallelism: int = 1, **kwargs: Any
) -> None:
- r"""
- Uploads a file to HDFS.
+ """
+ Upload a file to HDFS.
:param source: Local path to file or folder.
- If it's a folder, all the files inside of it will be uploaded.
+ If it's a folder, all the files inside it will be uploaded.
.. note:: This implies that folders empty of files will not be created remotely.
:param destination: PTarget HDFS path.
diff --git a/airflow/providers/apache/hive/hooks/hive.py b/airflow/providers/apache/hive/hooks/hive.py
index 8e81592f79..9ab3c8dd02 100644
--- a/airflow/providers/apache/hive/hooks/hive.py
+++ b/airflow/providers/apache/hive/hooks/hive.py
@@ -123,7 +123,7 @@ class HiveCliHook(BaseHook):
self.mapred_job_name = mapred_job_name
def _get_proxy_user(self) -> str:
- """This function set the proper proxy_user value in case the user overwrite the default."""
+ """Set the proper proxy_user value in case the user overwrite the default."""
conn = self.conn
proxy_user_value: str = conn.extra_dejson.get("proxy_user", "")
@@ -136,7 +136,7 @@ class HiveCliHook(BaseHook):
return proxy_user_value # The default proxy user (undefined)
def _prepare_cli_cmd(self) -> list[Any]:
- """This function creates the command list from available information."""
+ """Create the command list from available information."""
conn = self.conn
hive_bin = "hive"
cmd_extra = []
@@ -189,7 +189,7 @@ class HiveCliHook(BaseHook):
@staticmethod
def _prepare_hiveconf(d: dict[Any, Any]) -> list[Any]:
"""
- Prepares a list of hiveconf params from a dictionary of key value pairs.
+ Prepare a list of hiveconf params from a dictionary of key value pairs.
:param d:
@@ -345,7 +345,7 @@ class HiveCliHook(BaseHook):
**kwargs: Any,
) -> None:
"""
- Loads a pandas DataFrame into hive.
+ Load a pandas DataFrame into hive.
Hive data types will be inferred if not passed but column names will
not be sanitized.
@@ -416,7 +416,7 @@ class HiveCliHook(BaseHook):
tblproperties: dict[str, Any] | None = None,
) -> None:
"""
- Loads a local file into Hive.
+ Load a local file into Hive.
Note that the table generated in Hive uses ``STORED AS textfile``
which isn't the most efficient serialization format. If a
@@ -518,7 +518,7 @@ class HiveMetastoreHook(BaseHook):
self.__dict__["metastore"] = self.get_metastore_client()
def get_metastore_client(self) -> Any:
- """Returns a Hive thrift client."""
+ """Return a Hive thrift client."""
import hmsclient
from thrift.protocol import TBinaryProtocol
from thrift.transport import TSocket, TTransport
@@ -588,7 +588,7 @@ class HiveMetastoreHook(BaseHook):
def check_for_partition(self, schema: str, table: str, partition: str) -> bool:
"""
- Checks whether a partition exists.
+ Check whether a partition exists.
:param schema: Name of hive schema (database) @table belongs to
:param table: Name of hive table @partition belongs to
@@ -609,7 +609,7 @@ class HiveMetastoreHook(BaseHook):
def check_for_named_partition(self, schema: str, table: str, partition_name: str) -> Any:
"""
- Checks whether a partition with a given name exists.
+ Check whether a partition with a given name exists.
:param schema: Name of hive schema (database) @table belongs to
:param table: Name of hive table @partition belongs to
@@ -653,7 +653,7 @@ class HiveMetastoreHook(BaseHook):
def get_partitions(self, schema: str, table_name: str, partition_filter: str | None = None) -> list[Any]:
"""
- Returns a list of all partitions in a table.
+ Return a list of all partitions in a table.
Works only for tables with less than 32767 (java short max val).
For subpartitioned table, the number might easily exceed this.
@@ -691,7 +691,7 @@ class HiveMetastoreHook(BaseHook):
part_specs: list[Any], partition_key: str | None, filter_map: dict[str, Any] | None
) -> Any:
"""
- Helper method to get max partition of partitions with partition_key from part specs.
+ Get max partition of partitions with partition_key from part specs.
key:value pair in filter_map will be used to filter out partitions.
@@ -737,7 +737,7 @@ class HiveMetastoreHook(BaseHook):
filter_map: dict[Any, Any] | None = None,
) -> Any:
"""
- Returns the maximum value for all partitions with given field in a table.
+ Return the maximum value for all partitions with given field in a table.
If only one partition key exist in the table, the key will be used as field.
filter_map should be a partition_key:partition_value map and will be used to
@@ -840,7 +840,7 @@ class HiveServer2Hook(DbApiHook):
supports_autocommit = False
def get_conn(self, schema: str | None = None) -> Any:
- """Returns a Hive connection object."""
+ """Return a Hive connection object."""
username: str | None = None
password: str | None = None
diff --git a/airflow/providers/apache/hive/macros/hive.py b/airflow/providers/apache/hive/macros/hive.py
index 382b0b680b..2b083e20fa 100644
--- a/airflow/providers/apache/hive/macros/hive.py
+++ b/airflow/providers/apache/hive/macros/hive.py
@@ -24,7 +24,7 @@ def max_partition(
table, schema="default", field=None, filter_map=None, metastore_conn_id="metastore_default"
):
"""
- Gets the max partition for a table.
+ Get the max partition for a table.
:param schema: The hive schema the table lives in
:param table: The hive table you are interested in, supports the dot
@@ -52,7 +52,7 @@ def max_partition(
def _closest_date(target_dt, date_list, before_target=None) -> datetime.date | None:
"""
- This function finds the date in a list closest to the target date.
+ Find the date in a list closest to the target date.
An optional parameter can be given to get the closest before or after.
@@ -76,7 +76,7 @@ def closest_ds_partition(
table, ds, before=True, schema="default", metastore_conn_id="metastore_default"
) -> str | None:
"""
- This function finds the date in a list closest to the target date.
+ Find the date in a list closest to the target date.
An optional parameter can be given to get the closest before or after.
diff --git a/airflow/providers/apache/hive/transfers/mssql_to_hive.py b/airflow/providers/apache/hive/transfers/mssql_to_hive.py
index d803911956..6db76b0d93 100644
--- a/airflow/providers/apache/hive/transfers/mssql_to_hive.py
+++ b/airflow/providers/apache/hive/transfers/mssql_to_hive.py
@@ -101,7 +101,7 @@ class MsSqlToHiveOperator(BaseOperator):
@classmethod
def type_map(cls, mssql_type: int) -> str:
- """Maps MsSQL type to Hive type."""
+ """Map MsSQL type to Hive type."""
map_dict = {
pymssql.BINARY.value: "INT",
pymssql.DECIMAL.value: "FLOAT",
diff --git a/airflow/providers/apache/hive/transfers/mysql_to_hive.py b/airflow/providers/apache/hive/transfers/mysql_to_hive.py
index f7ad8acaa3..4c3b3cc5bd 100644
--- a/airflow/providers/apache/hive/transfers/mysql_to_hive.py
+++ b/airflow/providers/apache/hive/transfers/mysql_to_hive.py
@@ -114,7 +114,7 @@ class MySqlToHiveOperator(BaseOperator):
@classmethod
def type_map(cls, mysql_type: int) -> str:
- """Maps MySQL type to Hive type."""
+ """Map MySQL type to Hive type."""
types = MySQLdb.constants.FIELD_TYPE
type_map = {
types.BIT: "INT",
diff --git a/airflow/providers/apache/kafka/hooks/base.py b/airflow/providers/apache/kafka/hooks/base.py
index cc6f739ab9..d2ba9cb057 100644
--- a/airflow/providers/apache/kafka/hooks/base.py
+++ b/airflow/providers/apache/kafka/hooks/base.py
@@ -44,7 +44,7 @@ class KafkaBaseHook(BaseHook):
@staticmethod
def get_ui_field_behaviour() -> dict[str, Any]:
- """Returns custom field behaviour."""
+ """Return custom field behaviour."""
return {
"hidden_fields": ["schema", "login", "password", "port", "host"],
"relabeling": {"extra": "Config Dict"},
diff --git a/airflow/providers/apache/kafka/hooks/client.py b/airflow/providers/apache/kafka/hooks/client.py
index b8379c983e..fa4dff1d69 100644
--- a/airflow/providers/apache/kafka/hooks/client.py
+++ b/airflow/providers/apache/kafka/hooks/client.py
@@ -41,7 +41,7 @@ class KafkaAdminClientHook(KafkaBaseHook):
self,
topics: Sequence[Sequence[Any]],
) -> None:
- """Creates a topic.
+ """Create a topic.
:param topics: a list of topics to create including the number of partitions for the topic
and the replication factor. Format: [ ("topic_name", number of partitions, replication factor)]
diff --git a/airflow/providers/apache/kafka/hooks/consume.py b/airflow/providers/apache/kafka/hooks/consume.py
index d64859ffb1..5d9d7e421f 100644
--- a/airflow/providers/apache/kafka/hooks/consume.py
+++ b/airflow/providers/apache/kafka/hooks/consume.py
@@ -40,7 +40,7 @@ class KafkaConsumerHook(KafkaBaseHook):
return Consumer(config)
def get_consumer(self) -> Consumer:
- """Returns a Consumer that has been subscribed to topics."""
+ """Return a Consumer that has been subscribed to topics."""
consumer = self.get_conn
consumer.subscribe(self.topics)
diff --git a/airflow/providers/apache/kafka/hooks/produce.py b/airflow/providers/apache/kafka/hooks/produce.py
index 79d648b5e1..8675e228c3 100644
--- a/airflow/providers/apache/kafka/hooks/produce.py
+++ b/airflow/providers/apache/kafka/hooks/produce.py
@@ -35,7 +35,7 @@ class KafkaProducerHook(KafkaBaseHook):
return Producer(config)
def get_producer(self) -> Producer:
- """Returns a producer object for sending messages to Kafka."""
+ """Return a producer object for sending messages to Kafka."""
producer = self.get_conn
self.log.info("Producer %s", producer)
diff --git a/airflow/providers/apache/livy/hooks/livy.py b/airflow/providers/apache/livy/hooks/livy.py
index 151d03065e..ed4614cb28 100644
--- a/airflow/providers/apache/livy/hooks/livy.py
+++ b/airflow/providers/apache/livy/hooks/livy.py
@@ -92,7 +92,7 @@ class LivyHook(HttpHook, LoggingMixin):
def get_conn(self, headers: dict[str, Any] | None = None) -> Any:
"""
- Returns http session for use with requests.
+ Return http session for use with requests.
:param headers: additional headers to be passed through as a dictionary
:return: requests session
@@ -111,7 +111,7 @@ class LivyHook(HttpHook, LoggingMixin):
retry_args: dict[str, Any] | None = None,
) -> Any:
"""
- Wrapper for HttpHook, allows to change method on the same HttpHook.
+ Wrap HttpHook; allows to change method on the same HttpHook.
:param method: http method
:param endpoint: endpoint
@@ -254,7 +254,7 @@ class LivyHook(HttpHook, LoggingMixin):
def get_batch_logs(self, session_id: int | str, log_start_position, log_batch_size) -> dict:
"""
- Gets the session logs for a specified batch.
+ Get the session logs for a specified batch.
:param session_id: identifier of the batch sessions
:param log_start_position: Position from where to pull the logs
@@ -279,7 +279,7 @@ class LivyHook(HttpHook, LoggingMixin):
def dump_batch_logs(self, session_id: int | str) -> None:
"""
- Dumps the session logs for a specified batch.
+ Dump the session logs for a specified batch.
:param session_id: identifier of the batch sessions
:return: response body
@@ -498,7 +498,7 @@ class LivyAsyncHook(HttpAsyncHook, LoggingMixin):
extra_options: dict[str, Any] | None = None,
) -> Any:
"""
- Performs an asynchronous HTTP request call.
+ Perform an asynchronous HTTP request call.
:param endpoint: the endpoint to be called i.e. resource/v1/query?
:param data: payload to be uploaded or request parameters
@@ -591,7 +591,7 @@ class LivyAsyncHook(HttpAsyncHook, LoggingMixin):
headers: dict[str, Any] | None = None,
) -> Any:
"""
- Wrapper for HttpAsyncHook, allows to change method on the same HttpAsyncHook.
+ Wrap HttpAsyncHook; allows to change method on the same HttpAsyncHook.
:param method: http method
:param endpoint: endpoint
@@ -645,7 +645,7 @@ class LivyAsyncHook(HttpAsyncHook, LoggingMixin):
self, session_id: int | str, log_start_position: int, log_batch_size: int
) -> Any:
"""
- Gets the session logs for a specified batch asynchronously.
+ Get the session logs for a specified batch asynchronously.
:param session_id: identifier of the batch sessions
:param log_start_position: Position from where to pull the logs
@@ -662,7 +662,7 @@ class LivyAsyncHook(HttpAsyncHook, LoggingMixin):
async def dump_batch_logs(self, session_id: int | str) -> Any:
"""
- Dumps the session logs for a specified batch asynchronously.
+ Dump the session logs for a specified batch asynchronously.
:param session_id: identifier of the batch sessions
:return: response body
diff --git a/airflow/providers/apache/livy/operators/livy.py b/airflow/providers/apache/livy/operators/livy.py
index f5e519315f..046f33324e 100644
--- a/airflow/providers/apache/livy/operators/livy.py
+++ b/airflow/providers/apache/livy/operators/livy.py
@@ -202,7 +202,7 @@ class LivyOperator(BaseOperator):
def execute_complete(self, context: Context, event: dict[str, Any]) -> Any:
"""
- Callback for when the trigger fires - returns immediately.
+ Execute when the trigger fires - returns immediately.
Relies on trigger to throw an exception, otherwise it assumes execution was successful.
"""
diff --git a/airflow/providers/apache/livy/triggers/livy.py b/airflow/providers/apache/livy/triggers/livy.py
index c7b4e572bc..95ccc8577b 100644
--- a/airflow/providers/apache/livy/triggers/livy.py
+++ b/airflow/providers/apache/livy/triggers/livy.py
@@ -64,7 +64,7 @@ class LivyTrigger(BaseTrigger):
self._livy_hook_async = livy_hook_async
def serialize(self) -> tuple[str, dict[str, Any]]:
- """Serializes LivyTrigger arguments and classpath."""
+ """Serialize LivyTrigger arguments and classpath."""
return (
"airflow.providers.apache.livy.triggers.livy.LivyTrigger",
{
diff --git a/airflow/providers/apache/pinot/hooks/pinot.py b/airflow/providers/apache/pinot/hooks/pinot.py
index d9fd9044a2..0246a5d3a0 100644
--- a/airflow/providers/apache/pinot/hooks/pinot.py
+++ b/airflow/providers/apache/pinot/hooks/pinot.py
@@ -291,7 +291,7 @@ class PinotDbApiHook(DbApiHook):
self, sql: str | list[str], parameters: Iterable | Mapping[str, Any] | None = None, **kwargs
) -> Any:
"""
- Executes the sql and returns a set of records.
+ Execute the sql and returns a set of records.
:param sql: the sql statement to be executed (str) or a list of
sql statements to execute
@@ -303,7 +303,7 @@ class PinotDbApiHook(DbApiHook):
def get_first(self, sql: str | list[str], parameters: Iterable | Mapping[str, Any] | None = None) -> Any:
"""
- Executes the sql and returns the first resulting row.
+ Execute the sql and returns the first resulting row.
:param sql: the sql statement to be executed (str) or a list of
sql statements to execute
diff --git a/airflow/providers/apache/spark/hooks/spark_submit.py b/airflow/providers/apache/spark/hooks/spark_submit.py
index 22fd6d1436..97b1f4ebc9 100644
--- a/airflow/providers/apache/spark/hooks/spark_submit.py
+++ b/airflow/providers/apache/spark/hooks/spark_submit.py
@@ -87,7 +87,7 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
@staticmethod
def get_ui_field_behaviour() -> dict[str, Any]:
- """Returns custom field behaviour."""
+ """Return custom field behaviour."""
return {
"hidden_fields": ["schema", "login", "password"],
"relabeling": {},
@@ -445,7 +445,7 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
def _process_spark_submit_log(self, itr: Iterator[Any]) -> None:
"""
- Processes the log files and extracts useful information out of it.
+ Process the log files and extract useful information out of it.
If the deploy-mode is 'client', log the output of the submit command as those
are the output logs of the Spark worker directly.
@@ -492,7 +492,7 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
def _process_spark_status_log(self, itr: Iterator[Any]) -> None:
"""
- Parses the logs of the spark driver status query process.
+ Parse the logs of the spark driver status query process.
:param itr: An iterator which iterates over the input of the subprocess
"""
@@ -518,7 +518,7 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
def _start_driver_status_tracking(self) -> None:
"""
- Polls the driver based on self._driver_id to get the status.
+ Poll the driver based on self._driver_id to get the status.
Finish successfully when the status is FINISHED.
Finish failed when the status is ERROR/UNKNOWN/KILLED/FAILED.
diff --git a/airflow/providers/apache/sqoop/operators/sqoop.py b/airflow/providers/apache/sqoop/operators/sqoop.py
index c662afcc71..8aaae63dab 100644
--- a/airflow/providers/apache/sqoop/operators/sqoop.py
+++ b/airflow/providers/apache/sqoop/operators/sqoop.py
@@ -246,7 +246,7 @@ class SqoopOperator(BaseOperator):
os.killpg(os.getpgid(self.hook.sub_process_pid), signal.SIGTERM)
def _get_hook(self) -> SqoopHook:
- """Returns a SqoopHook instance."""
+ """Return a SqoopHook instance."""
# Add `create-hcatalog-table` to extra options if option passed to operator in case of `import`
# command. Similarly, if new parameters are added to the operator, you can pass them to
# `extra_options` so that you don't need to modify `SqoopHook` for each new parameter.
diff --git a/airflow/providers/apprise/hooks/apprise.py b/airflow/providers/apprise/hooks/apprise.py
index 5ab9e16da9..6abeaf658b 100644
--- a/airflow/providers/apprise/hooks/apprise.py
+++ b/airflow/providers/apprise/hooks/apprise.py
@@ -112,7 +112,7 @@ class AppriseHook(BaseHook):
@staticmethod
def get_connection_form_widgets() -> dict[str, Any]:
- """Returns connection widgets to add to connection form."""
+ """Return connection widgets to add to connection form."""
from flask_appbuilder.fieldwidgets import BS3PasswordFieldWidget
from flask_babel import lazy_gettext
from wtforms import PasswordField
diff --git a/airflow/providers/arangodb/hooks/arangodb.py b/airflow/providers/arangodb/hooks/arangodb.py
index 1495c598a5..07363e00f5 100644
--- a/airflow/providers/arangodb/hooks/arangodb.py
+++ b/airflow/providers/arangodb/hooks/arangodb.py
@@ -88,12 +88,12 @@ class ArangoDBHook(BaseHook):
return self._conn.password or ""
def get_conn(self) -> ArangoDBClient:
- """Function that initiates a new ArangoDB connection (cached)."""
+ """Initiate a new ArangoDB connection (cached)."""
return self.client
def query(self, query, **kwargs) -> Cursor:
"""
- Function to create an ArangoDB session and execute the AQL query in the session.
+ Create an ArangoDB session and execute the AQL query in the session.
:param query: AQL query
"""
diff --git a/airflow/providers/asana/hooks/asana.py b/airflow/providers/asana/hooks/asana.py
index 6278288877..7c76090ee7 100644
--- a/airflow/providers/asana/hooks/asana.py
+++ b/airflow/providers/asana/hooks/asana.py
@@ -85,7 +85,7 @@ class AsanaHook(BaseHook):
@staticmethod
def get_connection_form_widgets() -> dict[str, Any]:
- """Returns connection widgets to add to connection form."""
+ """Return connection widgets to add to connection form."""
from flask_appbuilder.fieldwidgets import BS3TextFieldWidget
from flask_babel import lazy_gettext
from wtforms import StringField
@@ -98,7 +98,7 @@ class AsanaHook(BaseHook):
@staticmethod
@_ensure_prefixes(conn_type="asana")
def get_ui_field_behaviour() -> dict[str, Any]:
- """Returns custom field behaviour."""
+ """Return custom field behaviour."""
return {
"hidden_fields": ["port", "host", "login", "schema"],
"relabeling": {},
@@ -111,7 +111,7 @@ class AsanaHook(BaseHook):
@cached_property
def client(self) -> Client:
- """Instantiates python-asana Client."""
+ """Instantiate python-asana Client."""
if not self.connection.password:
raise ValueError(
"Asana connection password must contain a personal access token: "
@@ -168,7 +168,7 @@ class AsanaHook(BaseHook):
def delete_task(self, task_id: str) -> dict:
"""
- Deletes an Asana task.
+ Delete an Asana task.
:param task_id: Asana GID of the task to delete
:return: A dict containing the response from Asana
@@ -182,7 +182,7 @@ class AsanaHook(BaseHook):
def find_task(self, params: dict | None) -> list:
"""
- Retrieves a list of Asana tasks that match search parameters.
+ Retrieve a list of Asana tasks that match search parameters.
:param params: Attributes that matching tasks should have. For a list of possible parameters,
see https://developers.asana.com/docs/get-multiple-tasks
@@ -231,7 +231,7 @@ class AsanaHook(BaseHook):
def update_task(self, task_id: str, params: dict) -> dict:
"""
- Updates an existing Asana task.
+ Update an existing Asana task.
:param task_id: Asana GID of task to update
:param params: New values of the task's attributes. For a list of possible parameters, see
@@ -243,7 +243,7 @@ class AsanaHook(BaseHook):
def create_project(self, params: dict) -> dict:
"""
- Creates a new project.
+ Create a new project.
:param params: Attributes that the new project should have. See
https://developers.asana.com/docs/create-a-project#create-a-project-parameters
@@ -283,7 +283,7 @@ class AsanaHook(BaseHook):
def find_project(self, params: dict) -> list:
"""
- Retrieves a list of Asana projects that match search parameters.
+ Retrieve a list of Asana projects that match search parameters.
:param params: Attributes which matching projects should have. See
https://developers.asana.com/docs/get-multiple-projects
@@ -296,7 +296,7 @@ class AsanaHook(BaseHook):
def update_project(self, project_id: str, params: dict) -> dict:
"""
- Updates an existing project.
+ Update an existing project.
:param project_id: Asana GID of the project to update
:param params: New attributes that the project should have. See
@@ -309,7 +309,7 @@ class AsanaHook(BaseHook):
def delete_project(self, project_id: str) -> dict:
"""
- Deletes a project.
+ Delete a project.
:param project_id: Asana GID of the project to delete
:return: A dict containing the response from Asana