Fix Sphinx Issues with Docstrings (#14968)

This PR fix some typos and issues were the doctrings do not adhere to sphinx standards in the following modules:
- AWSDataSyncOperator
- S3KeySizeSensor
- DatadogHook
- DatadogSensor
- ComputeEngineSSHHook
- CloudDataCatalogLookupEntryOperator
- GoogleDisplayVideo360UploadLineItemsOperator
- GoogleDisplayVideo360CreateSDFDownloadTaskOperator
- GoogleDisplayVideo360SDFtoGCSOperator
- AzureBatchOperator
- AzureCosmosDocumentSensor
- SingularityOperator
- SlackAPIOperator
- DataprocCreateMapReduceJobOperator
- DataprocCreateSparkJobOperator
- DataprocCreatePysparkJobOperator

Author: sunkickr

airflow/providers/amazon/aws/operators/datasync.py

@@ -42,48 +42,61 @@ class AWSDataSyncOperator(BaseOperator):
         environment. The default behavior is to create a new Task if there are 0, or
         execute the Task if there was 1 Task, or fail if there were many Tasks.
 
-    :param str aws_conn_id: AWS connection to use.
-    :param int wait_interval_seconds: Time to wait between two
+    :param aws_conn_id: AWS connection to use.
+    :type aws_conn_id: str
+    :param wait_interval_seconds: Time to wait between two
         consecutive calls to check TaskExecution status.
-    :param str task_arn: AWS DataSync TaskArn to use. If None, then this operator will
+    :type wait_interval_seconds: int
+    :param task_arn: AWS DataSync TaskArn to use. If None, then this operator will
         attempt to either search for an existing Task or attempt to create a new Task.
-    :param str source_location_uri: Source location URI to search for. All DataSync
+    :type task_arn: str
+    :param source_location_uri: Source location URI to search for. All DataSync
         Tasks with a LocationArn with this URI will be considered.
         Example: ``smb://server/subdir``
-    :param str destination_location_uri: Destination location URI to search for.
+    :type source_location_uri: str
+    :param destination_location_uri: Destination location URI to search for.
         All DataSync Tasks with a LocationArn with this URI will be considered.
         Example: ``s3://airflow_bucket/stuff``
-    :param bool allow_random_task_choice: If multiple Tasks match, one must be chosen to
+    :type destination_location_uri: str
+    :param allow_random_task_choice: If multiple Tasks match, one must be chosen to
         execute. If allow_random_task_choice is True then a random one is chosen.
-    :param bool allow_random_location_choice: If multiple Locations match, one must be chosen
+    :type allow_random_task_choice: bool
+    :param allow_random_location_choice: If multiple Locations match, one must be chosen
         when creating a task. If allow_random_location_choice is True then a random one is chosen.
-    :param dict create_task_kwargs: If no suitable TaskArn is identified,
+    :type allow_random_location_choice: bool
+    :param create_task_kwargs: If no suitable TaskArn is identified,
         it will be created if ``create_task_kwargs`` is defined.
         ``create_task_kwargs`` is then used internally like this:
         ``boto3.create_task(**create_task_kwargs)``
         Example:  ``{'Name': 'xyz', 'Options': ..., 'Excludes': ..., 'Tags': ...}``
-    :param dict create_source_location_kwargs: If no suitable LocationArn is found,
+    :type create_task_kwargs: dict
+    :param create_source_location_kwargs: If no suitable LocationArn is found,
         a Location will be created if ``create_source_location_kwargs`` is defined.
         ``create_source_location_kwargs`` is then used internally like this:
         ``boto3.create_location_xyz(**create_source_location_kwargs)``
         The xyz is determined from the prefix of source_location_uri, eg ``smb:/...`` or ``s3:/...``
         Example:  ``{'Subdirectory': ..., 'ServerHostname': ..., ...}``
-    :param dict create_destination_location_kwargs: If no suitable LocationArn is found,
+    :type create_source_location_kwargs: dict
+    :param create_destination_location_kwargs: If no suitable LocationArn is found,
         a Location will be created if ``create_destination_location_kwargs`` is defined.
         ``create_destination_location_kwargs`` is used internally like this:
         ``boto3.create_location_xyz(**create_destination_location_kwargs)``
         The xyz is determined from the prefix of destination_location_uri, eg ``smb:/...` or ``s3:/...``
         Example:  ``{'S3BucketArn': ..., 'S3Config': {'BucketAccessRoleArn': ...}, ...}``
-    :param dict update_task_kwargs:  If a suitable TaskArn is found or created,
+    :type create_destination_location_kwargs: dict
+    :param update_task_kwargs:  If a suitable TaskArn is found or created,
         it will be updated if ``update_task_kwargs`` is defined.
         ``update_task_kwargs`` is used internally like this:
         ``boto3.update_task(TaskArn=task_arn, **update_task_kwargs)``
         Example:  ``{'Name': 'xyz', 'Options': ..., 'Excludes': ...}``
-    :param dict task_execution_kwargs: Additional kwargs passed directly when starting the
+    :type update_task_kwargs: dict
+    :param task_execution_kwargs: Additional kwargs passed directly when starting the
         Task execution, used internally like this:
         ``boto3.start_task_execution(TaskArn=task_arn, **task_execution_kwargs)``
-    :param bool delete_task_after_execution: If True then the TaskArn which was executed
+    :type task_execution_kwargs: dict
+    :param  delete_task_after_execution: If True then the TaskArn which was executed
         will be deleted from AWS DataSync on successful completion.
+    :type delete_task_after_execution: bool
     :raises AirflowException: If ``task_arn`` was not specified, or if
         either ``source_location_uri`` or ``destination_location_uri`` were
         not specified.

airflow/providers/amazon/aws/sensors/s3_key.py

@@ -148,6 +148,7 @@ class S3KeySizeSensor(S3KeySensor):
 
             def check_fn(self, data: List) -> bool:
                 return any(f.get('Size', 0) > 1048576 for f in data if isinstance(f, dict))
+    :type check_fn: Optional[Callable[..., bool]]
     """
 
     @apply_defaults

airflow/providers/datadog/hooks/datadog.py

@@ -36,7 +36,7 @@ class DatadogHook(BaseHook, LoggingMixin):
     Airflow runs.
 
     :param datadog_conn_id: The connection to datadog, containing metadata for api keys.
-    :param datadog_conn_id: str
+    :type datadog_conn_id: str
     """
 
     def __init__(self, datadog_conn_id: str = 'datadog_default') -> None:

airflow/providers/datadog/sensors/datadog.py

@@ -34,7 +34,23 @@ class DatadogSensor(BaseSensorOperator):
     Airflow runs.
 
     :param datadog_conn_id: The connection to datadog, containing metadata for api keys.
-    :param datadog_conn_id: str
+    :type datadog_conn_id: str
+    :param from_seconds_ago: POSIX timestamp start (default 3600).
+    :type from_seconds_ago: int
+    :param up_to_seconds_from_now: POSIX timestamp end (default 0).
+    :type up_to_seconds_from_now: int
+    :param priority: Priority of your events, either low or normal.
+    :type priority: Optional[str]
+    :param sources: A comma separated list indicating what tags, if any,
+        should be used to filter the list of monitors by scope
+    :type sources: Optional[str]
+    :param tags: Get datadog events from specific sources.
+    :type tags: Optional[List[str]]
+    :param response_check: A check against the ‘requests’ response object. The callable takes
+        the response object as the first positional argument and optionally any number of
+        keyword arguments available in the context dictionary. It should return True for
+        ‘pass’ and False otherwise.
+    :param response_check: Optional[Callable[[Dict[str, Any]], bool]]
     """
 
     ui_color = '#66c3dd'

airflow/providers/google/cloud/hooks/compute_ssh.py

@@ -83,6 +83,7 @@ class ComputeEngineSSHHook(SSHHook):
     :type use_internal_ip: bool
     :param use_oslogin: Whether to manage keys using OsLogin API. If false,
         keys are managed using instance metadata
+    :type use_oslogin: bool
     :param expire_time: The maximum amount of time in seconds before the private key expires
     :type expire_time: int
     :param gcp_conn_id: The connection id to use when fetching connection information

airflow/providers/google/marketing_platform/operators/display_video.py

@@ -505,7 +505,7 @@ class GoogleDisplayVideo360UploadLineItemsOperator(BaseOperator):
     :param filename: The filename to fetch.
     :type filename: str,
     :param dry_run: Upload status without actually persisting the line items.
-    :type filename: str,
+    :type dry_run: str,
     """
 
     template_fields = (
@@ -572,7 +572,7 @@ class GoogleDisplayVideo360CreateSDFDownloadTaskOperator(BaseOperator):
         Check also the official API docs:
         `https://developers.google.com/display-video/api/reference/rest`
 
-    :param version: The SDF version of the downloaded file..
+    :param version: The SDF version of the downloaded file.
     :type version: str
     :param partner_id: The ID of the partner to download SDF for.
     :type partner_id: str
@@ -650,7 +650,7 @@ class GoogleDisplayVideo360SDFtoGCSOperator(BaseOperator):
         Check also the official API docs:
         `https://developers.google.com/display-video/api/reference/rest`
 
-    :param version: The SDF version of the downloaded file..
+    :param version: The SDF version of the downloaded file.
     :type version: str
     :param partner_id: The ID of the partner to download SDF for.
     :type partner_id: str

airflow/providers/microsoft/azure/operators/azure_batch.py

@@ -33,129 +33,94 @@ class AzureBatchOperator(BaseOperator):
 
     :param batch_pool_id: A string that uniquely identifies the Pool within the Account.
     :type batch_pool_id: str
-
     :param batch_pool_vm_size: The size of virtual machines in the Pool
     :type batch_pool_vm_size: str
-
     :param batch_job_id: A string that uniquely identifies the Job within the Account.
     :type batch_job_id: str
-
     :param batch_task_command_line: The command line of the Task
-    :type batch_command_line: str
-
+    :type batch_task_command_line: str
     :param batch_task_id: A string that uniquely identifies the task within the Job.
     :type batch_task_id: str
-
     :param batch_pool_display_name: The display name for the Pool.
         The display name need not be unique
     :type batch_pool_display_name: Optional[str]
-
     :param batch_job_display_name: The display name for the Job.
         The display name need not be unique
     :type batch_job_display_name: Optional[str]
-
     :param batch_job_manager_task: Details of a Job Manager Task to be launched when the Job is started.
-    :type job_manager_task: Optional[batch_models.JobManagerTask]
-
+    :type batch_job_manager_task: Optional[batch_models.JobManagerTask]
     :param batch_job_preparation_task: The Job Preparation Task. If set, the Batch service will
         run the Job Preparation Task on a Node before starting any Tasks of that
         Job on that Compute Node. Required if batch_job_release_task is set.
     :type batch_job_preparation_task: Optional[batch_models.JobPreparationTask]
-
     :param batch_job_release_task: The Job Release Task. Use to undo changes to Compute Nodes
         made by the Job Preparation Task
     :type batch_job_release_task: Optional[batch_models.JobReleaseTask]
-
     :param batch_task_display_name:  The display name for the task.
         The display name need not be unique
     :type batch_task_display_name: Optional[str]
-
     :param batch_task_container_settings: The settings for the container under which the Task runs
     :type batch_task_container_settings: Optional[batch_models.TaskContainerSettings]
-
     :param batch_start_task: A Task specified to run on each Compute Node as it joins the Pool.
         The Task runs when the Compute Node is added to the Pool or
         when the Compute Node is restarted.
     :type batch_start_task: Optional[batch_models.StartTask]
-
     :param batch_max_retries: The number of times to retry this batch operation before it's
         considered a failed operation. Default is 3
     :type batch_max_retries: int
-
     :param batch_task_resource_files: A list of files that the Batch service will
         download to the Compute Node before running the command line.
     :type batch_task_resource_files: Optional[List[batch_models.ResourceFile]]
-
     :param batch_task_output_files: A list of files that the Batch service will upload
         from the Compute Node after running the command line.
     :type batch_task_output_files: Optional[List[batch_models.OutputFile]]
-
     :param batch_task_user_identity: The user identity under which the Task runs.
         If omitted, the Task runs as a non-administrative user unique to the Task.
     :type batch_task_user_identity: Optional[batch_models.UserIdentity]
-
     :param target_low_priority_nodes: The desired number of low-priority Compute Nodes in the Pool.
         This property must not be specified if enable_auto_scale is set to true.
     :type target_low_priority_nodes: Optional[int]
-
     :param target_dedicated_nodes: The desired number of dedicated Compute Nodes in the Pool.
         This property must not be specified if enable_auto_scale is set to true.
     :type target_dedicated_nodes: Optional[int]
-
     :param enable_auto_scale: Whether the Pool size should automatically adjust over time. Default is false
     :type enable_auto_scale: bool
-
     :param auto_scale_formula: A formula for the desired number of Compute Nodes in the Pool.
         This property must not be specified if enableAutoScale is set to false.
         It is required if enableAutoScale is set to true.
     :type auto_scale_formula: Optional[str]
-
     :param azure_batch_conn_id: The connection id of Azure batch service
     :type azure_batch_conn_id: str
-
     :param use_latest_verified_vm_image_and_sku: Whether to use the latest verified virtual
         machine image and sku in the batch account. Default is false.
     :type use_latest_verified_vm_image_and_sku: bool
-
     :param vm_publisher: The publisher of the Azure Virtual Machines Marketplace Image.
         For example, Canonical or MicrosoftWindowsServer. Required if
         use_latest_image_and_sku is set to True
     :type vm_publisher: Optional[str]
-
     :param vm_offer: The offer type of the Azure Virtual Machines Marketplace Image.
         For example, UbuntuServer or WindowsServer. Required if
         use_latest_image_and_sku is set to True
     :type vm_offer: Optional[str]
-
     :param sku_starts_with: The starting string of the Virtual Machine SKU. Required if
         use_latest_image_and_sku is set to True
     :type sku_starts_with: Optional[str]
-
     :param vm_sku: The name of the virtual machine sku to use
     :type vm_sku: Optional[str]
-
     :param vm_version: The version of the virtual machine
     :param vm_version: Optional[str]
-
     :param vm_node_agent_sku_id: The node agent sku id of the virtual machine
     :type vm_node_agent_sku_id: Optional[str]
-
     :param os_family: The Azure Guest OS family to be installed on the virtual machines in the Pool.
     :type os_family: Optional[str]
-
     :param os_version: The OS family version
     :type os_version: Optional[str]
-
     :param timeout: The amount of time to wait for the job to complete in minutes. Default is 25
     :type timeout: int
-
     :param should_delete_job: Whether to delete job after execution. Default is False
     :type should_delete_job: bool
-
     :param should_delete_pool: Whether to delete pool after execution of jobs. Default is False
     :type should_delete_pool: bool
-
-
     """
 
     template_fields = (

airflow/providers/microsoft/azure/sensors/azure_cosmos.py

@@ -26,18 +26,21 @@ class AzureCosmosDocumentSensor(BaseSensorOperator):
     Checks for the existence of a document which
     matches the given query in CosmosDB. Example:
 
-    >>> azure_cosmos_sensor = AzureCosmosDocumentSensor(database_name="somedatabase_name",
-    ...                            collection_name="somecollection_name",
-    ...                            document_id="unique-doc-id",
-    ...                            azure_cosmos_conn_id="azure_cosmos_default",
-    ...                            task_id="azure_cosmos_sensor")
+    .. code-block::
+
+    azure_cosmos_sensor = AzureCosmosDocumentSensor(
+        database_name="somedatabase_name",
+        collection_name="somecollection_name",
+        document_id="unique-doc-id",
+        azure_cosmos_conn_id="azure_cosmos_default",
+        task_id="azure_cosmos_sensor")
 
     :param database_name: Target CosmosDB database_name.
     :type database_name: str
     :param collection_name: Target CosmosDB collection_name.
     :type collection_name: str
     :param document_id: The ID of the target document.
-    :type query: str
+    :type document_id: str
     :param azure_cosmos_conn_id: Reference to the Azure CosmosDB connection.
     :type azure_cosmos_conn_id: str
     """

airflow/providers/singularity/operators/singularity.py

@@ -39,13 +39,13 @@ class SingularityOperator(BaseOperator):
 
     :param image: Singularity image or URI from which to create the container.
     :type image: str
-    :param auto_remove: Delete the container when the process exits
-                        The default is False.
+    :param auto_remove: Delete the container when the process exits.
+        The default is False.
     :type auto_remove: bool
     :param command: Command to be run in the container. (templated)
     :type command: str or list
-    :param start_command: start command to pass to the container instance
-    :type start_command: string or list
+    :param start_command: Start command to pass to the container instance.
+    :type start_command: str or list
     :param environment: Environment variables to set in the container. (templated)
     :type environment: dict
     :param working_dir: Set a working directory for the instance.
@@ -54,10 +54,11 @@ class SingularityOperator(BaseOperator):
     :type force_pull: bool
     :param volumes: List of volumes to mount into the container, e.g.
         ``['/host/path:/container/path', '/host/path2:/container/path2']``.
-    :param options: other flags (list) to provide to the instance start
+    :type volumes: Optional[List[str]]
+    :param options: Other flags (list) to provide to the instance start.
     :type options: list
     :param working_dir: Working directory to
-        set on the container (equivalent to the -w switch the docker client)
+        set on the container (equivalent to the -w switch the docker client).
     :type working_dir: str
     """
 

airflow/providers/slack/operators/slack.py

@@ -28,7 +28,7 @@ class SlackAPIOperator(BaseOperator):
     """
     Base Slack Operator
     The SlackAPIPostOperator is derived from this operator.
-    In the future additional Slack API Operators will be derived from this class as well
+    In the future additional Slack API Operators will be derived from this class as well.
     Only one of `slack_conn_id` and `token` is required.
 
     :param slack_conn_id: Slack connection ID which its password is Slack API token. Optional
@@ -40,7 +40,7 @@ class SlackAPIOperator(BaseOperator):
     :param api_params: API Method call parameters (https://api.slack.com/methods). Optional
     :type api_params: dict
     :param client_args: Slack Hook parameters. Optional. Check airflow.providers.slack.hooks.SlackHook
-    :type api_params: dict
+    :type client_args: dict
     """
 
     @apply_defaults