Skip to content

Commit

Permalink
Microsoft provider docstring improvements (#31708)
Browse files Browse the repository at this point in the history 
* Reformat ASB hook docstrings

* Reformat Data Lake hook docstrings

* Clean up ASB operator docstrings

* Reformat Cosmos operator docstring

* Rewrite Data Factory and WASB sensor docstrings

* Reformat Data Factory trigger docstrings

* Fix periods missed by misconfigured merge

Co-authored-by: Vincent <97131062+vincbeck@users.noreply.github.com>

---------

Co-authored-by: Vincent <97131062+vincbeck@users.noreply.github.com>
  • Loading branch information
@uranusjr @vincbeck
uranusjr and vincbeck committed Jun 20, 2023
1 parent f0b91ac commit 4266a54
Show file tree
Hide file tree
Showing 7 changed files with 83 additions and 108 deletions.
35 changes: 18 additions & 17 deletions airflow/providers/microsoft/azure/hooks/asb.py
View file
Expand Up @@ -57,17 +57,17 @@ def get_conn(self):


class AdminClientHook(BaseAzureServiceBusHook):
"""
Interacts with ServiceBusAdministrationClient client
to create, update, list, and delete resources of a
Service Bus namespace. This hook uses the same Azure Service Bus client connection inherited
from the base class.
"""Interact with the ServiceBusAdministrationClient.
This can create, update, list, and delete resources of a Service Bus
namespace. This hook uses the same Azure Service Bus client connection
inherited from the base class.
"""

def get_conn(self) -> ServiceBusAdministrationClient:
"""
Create and returns ServiceBusAdministrationClient by using the connection
string in connection details.
"""Create a ServiceBusAdministrationClient instance.
This uses the connection string in connection details.
"""
conn = self.get_connection(self.conn_id)

Expand Down Expand Up @@ -134,9 +134,9 @@ def delete_subscription(self, subscription_name: str, topic_name: str) -> None:


class MessageHook(BaseAzureServiceBusHook):
"""
Interacts with ServiceBusClient and acts as a high level interface
for getting ServiceBusSender and ServiceBusReceiver.
"""Interact with ServiceBusClient.
This acts as a high level interface for getting ServiceBusSender and ServiceBusReceiver.
"""

def get_conn(self) -> ServiceBusClient:
Expand All @@ -148,9 +148,9 @@ def get_conn(self) -> ServiceBusClient:
return ServiceBusClient.from_connection_string(conn_str=connection_string, logging_enable=True)

def send_message(self, queue_name: str, messages: str | list[str], batch_message_flag: bool = False):
"""
By using ServiceBusClient Send message(s) to a Service Bus Queue. By using
batch_message_flag it enables and send message as batch message.
"""Use ServiceBusClient Send to send message(s) to a Service Bus Queue.
By using ``batch_message_flag``, it enables and send message as batch message.
:param queue_name: The name of the queue or a QueueProperties with name.
:param messages: Message which needs to be sent to the queue. It can be string or list of string.
Expand Down Expand Up @@ -220,9 +220,10 @@ def receive_subscription_message(
max_message_count: int | None,
max_wait_time: float | None,
):
"""
Receive a batch of subscription message at once. This approach is optimal if you wish
to process multiple messages simultaneously, or perform an ad-hoc receive as a single call.
"""Receive a batch of subscription message at once.
This approach is optimal if you wish to process multiple messages
simultaneously, or perform an ad-hoc receive as a single call.
:param subscription_name: The subscription name that will own the rule in topic
:param topic_name: The topic that will own the subscription rule.
Expand Down
84 changes: 34 additions & 50 deletions airflow/providers/microsoft/azure/hooks/data_lake.py
View file
Expand Up @@ -37,21 +37,20 @@


class AzureDataLakeHook(BaseHook):
"""
This module contains integration with Azure Data Lake.
AzureDataLakeHook communicates via a REST API compatible with WebHDFS. Make sure that a
Airflow connection of type `azure_data_lake` exists. Authorization can be done by supplying a
login (=Client ID), password (=Client Secret) and extra fields tenant (Tenant) and account_name
(Account Name)(see connection `azure_data_lake_default` for an example).
"""Integration with Azure Data Lake.
Interacts with Azure Data Lake.
AzureDataLakeHook communicates via a REST API compatible with WebHDFS. Make
sure that a Airflow connection of type ``azure_data_lake`` exists.
Authorization can be done by supplying a *login* (=Client ID), *password*
(=Client Secret), and extra fields *tenant* (Tenant) and *account_name*
(Account Name). See connection ``azure_data_lake_default`` for an example.
Client ID and client secret should be in user and password parameters.
Client ID and secret should be in user and password parameters.
Tenant and account name should be extra field as
{"tenant": "<TENANT>", "account_name": "ACCOUNT_NAME"}.
``{"tenant": "<TENANT>", "account_name": "ACCOUNT_NAME"}``.
:param azure_data_lake_conn_id: Reference to the :ref:`Azure Data Lake connection<howto/connection:adl>`.
:param azure_data_lake_conn_id: Reference to
:ref:`Azure Data Lake connection<howto/connection:adl>`.
"""

conn_name_attr = "azure_data_lake_conn_id"
Expand Down Expand Up @@ -118,8 +117,7 @@ def get_conn(self) -> core.AzureDLFileSystem:
return self._conn

def check_for_file(self, file_path: str) -> bool:
"""
Check if a file exists on Azure Data Lake.
"""Check if a file exists on Azure Data Lake.
:param file_path: Path and name of the file.
:return: True if the file exists, False otherwise.
Expand All @@ -140,8 +138,7 @@ def upload_file(
blocksize: int = 4194304,
**kwargs,
) -> None:
"""
Upload a file to Azure Data Lake.
"""Upload a file to Azure Data Lake.
:param local_path: local path. Can be single file, directory (in which case,
upload recursively) or glob pattern. Recursive glob patterns using `**`
Expand Down Expand Up @@ -181,8 +178,7 @@ def download_file(
blocksize: int = 4194304,
**kwargs,
) -> None:
"""
Download a file from Azure Blob Storage.
"""Download a file from Azure Blob Storage.
:param local_path: local path. If downloading a single file, will write to this
specific file, unless it is an existing directory, in which case a file is
Expand Down Expand Up @@ -214,8 +210,7 @@ def download_file(
)

def list(self, path: str) -> list:
"""
List files in Azure Data Lake Storage.
"""List files in Azure Data Lake Storage.
:param path: full path/globstring to use to list files in ADLS
"""
Expand All @@ -225,8 +220,7 @@ def list(self, path: str) -> list:
return self.get_conn().walk(path)

def remove(self, path: str, recursive: bool = False, ignore_not_found: bool = True) -> None:
"""
Remove files in Azure Data Lake Storage.
"""Remove files in Azure Data Lake Storage.
:param path: A directory or file to remove in ADLS
:param recursive: Whether to loop into directories in the location and remove the files
Expand All @@ -242,13 +236,14 @@ def remove(self, path: str, recursive: bool = False, ignore_not_found: bool = Tr


class AzureDataLakeStorageV2Hook(BaseHook):
"""
This Hook interacts with ADLS gen2 storage account it mainly helps to create and manage
directories and files in storage accounts that have a hierarchical namespace. Using Adls_v2 connection
"""Interact with a ADLS gen2 storage account.
It mainly helps to create and manage directories and files in storage
accounts that have a hierarchical namespace. Using Adls_v2 connection
details create DataLakeServiceClient object.
Due to Wasb is marked as legacy and and retirement of the (ADLS1) it would be nice to
implement ADLS gen2 hook for interacting with the storage account.
Due to Wasb is marked as legacy and and retirement of the (ADLS1), it would
be nice to implement ADLS gen2 hook for interacting with the storage account.
.. seealso::
https://learn.microsoft.com/en-us/azure/storage/blobs/data-lake-storage-directory-file-acl-python
Expand Down Expand Up @@ -342,9 +337,9 @@ def _get_field(self, extra_dict, field_name):
return extra_dict.get(f"{prefix}{field_name}") or None

def create_file_system(self, file_system_name: str) -> None:
"""
A container acts as a file system for your files. Creates a new file system under
the specified account.
"""Create a new file system under the specified account.
A container acts as a file system for your files.
If the file system with the same name already exists, a ResourceExistsError will
be raised. This method returns a client with which to interact with the newly
Expand All @@ -360,8 +355,7 @@ def create_file_system(self, file_system_name: str) -> None:
raise

def get_file_system(self, file_system: FileSystemProperties | str) -> FileSystemClient:
"""
Get a client to interact with the specified file system.
"""Get a client to interact with the specified file system.
:param file_system: This can either be the name of the file system
or an instance of FileSystemProperties.
Expand All @@ -379,8 +373,7 @@ def get_file_system(self, file_system: FileSystemProperties | str) -> FileSystem
def create_directory(
self, file_system_name: FileSystemProperties | str, directory_name: str, **kwargs
) -> DataLakeDirectoryClient:
"""
Create a directory under the specified file system.
"""Create a directory under the specified file system.
:param file_system_name: Name of the file system or instance of FileSystemProperties.
:param directory_name: Name of the directory which needs to be created in the file system.
Expand All @@ -393,8 +386,7 @@ def get_directory_client(
file_system_name: FileSystemProperties | str,
directory_name: DirectoryProperties | str,
) -> DataLakeDirectoryClient:
"""
Get the specific directory under the specified file system.
"""Get the specific directory under the specified file system.
:param file_system_name: Name of the file system or instance of FileSystemProperties.
:param directory_name: Name of the directory or instance of DirectoryProperties which needs to be
Expand All @@ -413,8 +405,7 @@ def get_directory_client(
raise

def create_file(self, file_system_name: FileSystemProperties | str, file_name: str) -> DataLakeFileClient:
"""
Creates a file under the file system.
"""Create a file under the file system.
:param file_system_name: Name of the file system or instance of FileSystemProperties.
:param file_name: Name of the file which needs to be created in the file system.
Expand All @@ -430,8 +421,7 @@ def upload_file(
overwrite: bool = False,
**kwargs: Any,
) -> None:
"""
Create a file with data in the file system.
"""Create a file with data in the file system.
:param file_system_name: Name of the file system or instance of FileSystemProperties.
:param file_name: Name of the file to be created with name.
Expand All @@ -451,9 +441,7 @@ def upload_file_to_directory(
overwrite: bool = False,
**kwargs: Any,
) -> None:
"""
Create a new file and return the file client to be interacted with and then
upload data to a file.
"""Upload data to a file.
:param file_system_name: Name of the file system or instance of FileSystemProperties.
:param directory_name: Name of the directory.
Expand All @@ -469,8 +457,7 @@ def upload_file_to_directory(
def list_files_directory(
self, file_system_name: FileSystemProperties | str, directory_name: str
) -> list[str]:
"""
Get the list of files or directories under the specified file system.
"""List files or directories under the specified file system.
:param file_system_name: Name of the file system or instance of FileSystemProperties.
:param directory_name: Name of the directory.
Expand All @@ -484,8 +471,7 @@ def list_files_directory(
def list_file_system(
self, prefix: str | None = None, include_metadata: bool = False, **kwargs: Any
) -> list[str]:
"""
Get the list the file systems under the specified account.
"""List file systems under the specified account.
:param prefix:
Filters the results to return only file systems whose names
Expand All @@ -502,8 +488,7 @@ def list_file_system(
return file_system_list

def delete_file_system(self, file_system_name: FileSystemProperties | str) -> None:
"""
Deletes the file system.
"""Delete the file system.
:param file_system_name: Name of the file system or instance of FileSystemProperties.
"""
Expand All @@ -517,8 +502,7 @@ def delete_file_system(self, file_system_name: FileSystemProperties | str) -> No
raise

def delete_directory(self, file_system_name: FileSystemProperties | str, directory_name: str) -> None:
"""
Deletes specified directory in file system.
"""Delete the specified directory in a file system.
:param file_system_name: Name of the file system or instance of FileSystemProperties.
:param directory_name: Name of the directory.
Expand Down
37 changes: 11 additions & 26 deletions airflow/providers/microsoft/azure/operators/asb.py
View file
Expand Up @@ -29,8 +29,7 @@


class AzureServiceBusCreateQueueOperator(BaseOperator):
"""
Creates a Azure Service Bus queue under a Service Bus Namespace by using ServiceBusAdministrationClient.
"""Create a Azure Service Bus queue under a Service Bus Namespace.
.. seealso::
For more information on how to use this operator, take a look at the guide:
Expand Down Expand Up @@ -82,8 +81,7 @@ def execute(self, context: Context) -> None:


class AzureServiceBusSendMessageOperator(BaseOperator):
"""
Send Message or batch message to the Service Bus queue.
"""Send Message or batch message to the Service Bus queue.
.. seealso::
For more information on how to use this operator, take a look at the guide:
Expand Down Expand Up @@ -116,10 +114,7 @@ def __init__(
self.azure_service_bus_conn_id = azure_service_bus_conn_id

def execute(self, context: Context) -> None:
"""
Sends Message to the specific queue in Service Bus namespace, by
connecting to Service Bus client.
"""
"""Sends Message to the specific queue in Service Bus namespace."""
# Create the hook
hook = MessageHook(azure_service_bus_conn_id=self.azure_service_bus_conn_id)

Expand All @@ -128,8 +123,7 @@ def execute(self, context: Context) -> None:


class AzureServiceBusReceiveMessageOperator(BaseOperator):
"""
Receive a batch of messages at once in a specified Queue name.
"""Receive a batch of messages at once in a specified Queue name.
.. seealso::
For more information on how to use this operator, take a look at the guide:
Expand Down Expand Up @@ -175,8 +169,7 @@ def execute(self, context: Context) -> None:


class AzureServiceBusDeleteQueueOperator(BaseOperator):
"""
Deletes the Queue in the Azure Service Bus namespace.
"""Delete the Queue in the Azure Service Bus namespace.
.. seealso::
For more information on how to use this operator, take a look at the guide:
Expand Down Expand Up @@ -211,8 +204,7 @@ def execute(self, context: Context) -> None:


class AzureServiceBusTopicCreateOperator(BaseOperator):
"""
Create an Azure Service Bus Topic under a Service Bus Namespace by using ServiceBusAdministrationClient.
"""Create an Azure Service Bus Topic under a Service Bus Namespace.
.. seealso::
For more information on how to use this operator, take a look at the guide:
Expand Down Expand Up @@ -330,9 +322,7 @@ def execute(self, context: Context) -> str:


class AzureServiceBusSubscriptionCreateOperator(BaseOperator):
"""
Create an Azure Service Bus Topic Subscription under a Service Bus Namespace
by using ServiceBusAdministrationClient.
"""Create an Azure Service Bus Topic Subscription under a Service Bus Namespace.
.. seealso::
For more information on how to use this operator, take a look at the guide:
Expand Down Expand Up @@ -438,9 +428,7 @@ def execute(self, context: Context) -> None:


class AzureServiceBusUpdateSubscriptionOperator(BaseOperator):
"""
Update an Azure ServiceBus Topic Subscription under a ServiceBus Namespace
by using ServiceBusAdministrationClient.
"""Update an Azure ServiceBus Topic Subscription under a ServiceBus Namespace.
.. seealso::
For more information on how to use this operator, take a look at the guide:
Expand Down Expand Up @@ -499,8 +487,7 @@ def execute(self, context: Context) -> None:


class ASBReceiveSubscriptionMessageOperator(BaseOperator):
"""
Receive a Batch messages from a Service Bus Subscription under specific Topic.
"""Receive a Batch messages from a Service Bus Subscription under specific Topic.
.. seealso::
For more information on how to use this operator, take a look at the guide:
Expand Down Expand Up @@ -554,8 +541,7 @@ def execute(self, context: Context) -> None:


class AzureServiceBusSubscriptionDeleteOperator(BaseOperator):
"""
Deletes the topic subscription in the Azure ServiceBus namespace.
"""Delete the topic subscription in the Azure ServiceBus namespace.
.. seealso::
For more information on how to use this operator, take a look at the guide:
Expand Down Expand Up @@ -593,8 +579,7 @@ def execute(self, context: Context) -> None:


class AzureServiceBusTopicDeleteOperator(BaseOperator):
"""
Deletes the topic in the Azure Service Bus namespace.
"""Delete the topic in the Azure Service Bus namespace.
.. seealso::
For more information on how to use this operator, take a look at the guide:
Expand Down

0 comments on commit 4266a54

Please sign in to comment.