diff --git a/airflow/providers/google/cloud/hooks/automl.py b/airflow/providers/google/cloud/hooks/automl.py index dba14d720b12b..cada4eb511290 100644 --- a/airflow/providers/google/cloud/hooks/automl.py +++ b/airflow/providers/google/cloud/hooks/automl.py @@ -122,10 +122,10 @@ def create_model( retry: Retry | _MethodDefault = DEFAULT, ) -> Operation: """ - Creates a model_id. Returns a Model in the `response` field when it - completes. When you create a model, several model evaluations are - created for it: a global evaluation, and one evaluation for each - annotation spec. + Creates a model_id and returns a Model in the `response` field when it completes. + + When you create a model, several model evaluations are created for it: + a global evaluation, and one evaluation for each annotation spec. :param model: The model_id to create. If a dict is provided, it must be of the same form as the protobuf message `google.cloud.automl_v1beta1.types.Model` @@ -163,9 +163,10 @@ def batch_predict( metadata: Sequence[tuple[str, str]] = (), ) -> Operation: """ - Perform a batch prediction. Unlike the online `Predict`, batch - prediction result won't be immediately available in the response. - Instead, a long running operation object is returned. + Perform a batch prediction and returns a long-running operation object. + + Unlike the online `Predict`, batch prediction result won't be immediately + available in the response. Instead, a long-running operation object is returned. :param model_id: Name of the model_id requested to serve the batch prediction. :param input_config: Required. The input configuration for batch prediction. @@ -215,8 +216,7 @@ def predict( metadata: Sequence[tuple[str, str]] = (), ) -> PredictResponse: """ - Perform an online prediction. The prediction result will be directly - returned in the response. + Perform an online prediction and returns the prediction result in the response. :param model_id: Name of the model_id requested to serve the prediction. :param payload: Required. Payload to perform a prediction on. The payload must match the problem type @@ -485,7 +485,9 @@ def deploy_model( metadata: Sequence[tuple[str, str]] = (), ) -> Operation: """ - Deploys a model. If a model is already deployed, deploying it with the same parameters + Deploys a model. + + If a model is already deployed, deploying it with the same parameters has no effect. Deploying with different parameters (as e.g. changing node_number) will reset the deployment state without pausing the model_id's availability. diff --git a/airflow/providers/google/cloud/hooks/bigquery_dts.py b/airflow/providers/google/cloud/hooks/bigquery_dts.py index c2dff601d25e0..2d233591fde14 100644 --- a/airflow/providers/google/cloud/hooks/bigquery_dts.py +++ b/airflow/providers/google/cloud/hooks/bigquery_dts.py @@ -75,6 +75,8 @@ def __init__( @staticmethod def _disable_auto_scheduling(config: dict | TransferConfig) -> TransferConfig: """ + Create a transfer config with the automatic scheduling disabled. + In the case of Airflow, the customer needs to create a transfer config with the automatic scheduling disabled (UI, CLI or an Airflow operator) and then trigger a transfer run using a specialized Airflow operator that will @@ -195,10 +197,10 @@ def start_manual_transfer_runs( metadata: Sequence[tuple[str, str]] = (), ) -> StartManualTransferRunsResponse: """ - Start manual transfer runs to be executed now with schedule_time equal - to current time. The transfer runs can be created for a time range where - the run_time is between start_time (inclusive) and end_time - (exclusive), or for a specific run_time. + Start manual transfer runs to be executed now with schedule_time equal to current time. + + The transfer runs can be created for a time range where the run_time is between + start_time (inclusive) and end_time (exclusive), or for a specific run_time. :param transfer_config_id: Id of transfer config to be used. :param requested_time_range: Time range for the transfer runs that should be started. diff --git a/airflow/providers/google/cloud/hooks/bigtable.py b/airflow/providers/google/cloud/hooks/bigtable.py index c1e573a85a738..d01ab2d8ee1a7 100644 --- a/airflow/providers/google/cloud/hooks/bigtable.py +++ b/airflow/providers/google/cloud/hooks/bigtable.py @@ -69,8 +69,7 @@ def _get_client(self, project_id: str) -> Client: @GoogleBaseHook.fallback_to_default_project_id def get_instance(self, instance_id: str, project_id: str) -> Instance | None: """ - Retrieves and returns the specified Cloud Bigtable instance if it exists. - Otherwise, returns None. + Retrieves and returns the specified Cloud Bigtable instance if it exists, otherwise returns None. :param instance_id: The ID of the Cloud Bigtable instance. :param project_id: Optional, Google Cloud project ID where the @@ -86,6 +85,7 @@ def get_instance(self, instance_id: str, project_id: str) -> Instance | None: def delete_instance(self, instance_id: str, project_id: str) -> None: """ Deletes the specified Cloud Bigtable instance. + Raises google.api_core.exceptions.NotFound if the Cloud Bigtable instance does not exist. @@ -217,6 +217,7 @@ def create_table( ) -> None: """ Creates the specified Cloud Bigtable table. + Raises ``google.api_core.exceptions.AlreadyExists`` if the table exists. :param instance: The Cloud Bigtable instance that owns the table. @@ -238,6 +239,7 @@ def create_table( def delete_table(self, instance_id: str, table_id: str, project_id: str) -> None: """ Deletes the specified table in Cloud Bigtable. + Raises google.api_core.exceptions.NotFound if the table does not exist. :param instance_id: The ID of the Cloud Bigtable instance. @@ -256,6 +258,7 @@ def delete_table(self, instance_id: str, table_id: str, project_id: str) -> None def update_cluster(instance: Instance, cluster_id: str, nodes: int) -> None: """ Updates number of nodes in the specified Cloud Bigtable cluster. + Raises google.api_core.exceptions.NotFound if the cluster does not exist. :param instance: The Cloud Bigtable instance that owns the cluster. @@ -284,6 +287,7 @@ def get_column_families_for_table(instance: Instance, table_id: str) -> dict[str def get_cluster_states_for_table(instance: Instance, table_id: str) -> dict[str, ClusterState]: """ Fetches Cluster States for the specified table in Cloud Bigtable. + Raises google.api_core.exceptions.NotFound if the table does not exist. :param instance: The Cloud Bigtable instance that owns the table. diff --git a/airflow/providers/google/cloud/hooks/cloud_build.py b/airflow/providers/google/cloud/hooks/cloud_build.py index 0ac929a4ad2fb..1b7815d9035aa 100644 --- a/airflow/providers/google/cloud/hooks/cloud_build.py +++ b/airflow/providers/google/cloud/hooks/cloud_build.py @@ -502,8 +502,7 @@ def retry_build( location: str = "global", ) -> Build: """ - Creates a new build based on the specified build. This method creates a new build - using the original build request, which may or may not result in an identical build. + Create a new build using the original build request; may or may not result in an identical build. :param id_: Build ID of the original build. :param project_id: Optional, Google Cloud Project project_id where the function belongs. diff --git a/airflow/providers/google/cloud/hooks/cloud_composer.py b/airflow/providers/google/cloud/hooks/cloud_composer.py index 5c994b3a58501..3ad0ba1098ac3 100644 --- a/airflow/providers/google/cloud/hooks/cloud_composer.py +++ b/airflow/providers/google/cloud/hooks/cloud_composer.py @@ -156,6 +156,7 @@ def get_environment( ) -> Environment: """ Get an existing environment. + :param project_id: Required. The ID of the Google Cloud project that the service belongs to. :param region: Required. The ID of the Google Cloud region that the service belongs to. :param environment_id: Required. The ID of the Google Cloud environment that the service belongs to. diff --git a/airflow/providers/google/cloud/hooks/cloud_sql.py b/airflow/providers/google/cloud/hooks/cloud_sql.py index 304a6f88fa3d2..ebd03b69bbb68 100644 --- a/airflow/providers/google/cloud/hooks/cloud_sql.py +++ b/airflow/providers/google/cloud/hooks/cloud_sql.py @@ -304,8 +304,7 @@ def delete_database(self, instance: str, database: str, project_id: str) -> None @GoogleBaseHook.fallback_to_default_project_id def export_instance(self, instance: str, body: dict, project_id: str): """ - Exports data from a Cloud SQL instance to a Cloud Storage bucket as a SQL dump - or CSV file. + Exports data from a Cloud SQL instance to a Cloud Storage bucket as a SQL dump or CSV file. :param instance: Database instance ID of the Cloud SQL instance. This does not include the project ID. @@ -327,8 +326,7 @@ def export_instance(self, instance: str, body: dict, project_id: str): @GoogleBaseHook.fallback_to_default_project_id def import_instance(self, instance: str, body: dict, project_id: str) -> None: """ - Imports data into a Cloud SQL instance from a SQL dump or CSV file in - Cloud Storage. + Imports data into a Cloud SQL instance from a SQL dump or CSV file in Cloud Storage. :param instance: Database instance ID. This does not include the project ID. @@ -382,8 +380,7 @@ def _wait_for_operation_to_complete( self, project_id: str, operation_name: str, time_to_sleep: int = TIME_TO_SLEEP_IN_SECONDS ) -> None: """ - Waits for the named operation to complete - checks status of the - asynchronous call. + Waits for the named operation to complete - checks status of the asynchronous call. :param project_id: Project ID of the project that contains the instance. :param operation_name: Name of the operation. @@ -721,10 +718,11 @@ def get_socket_path(self) -> str: class CloudSQLDatabaseHook(BaseHook): - """Serves DB connection configuration for Google Cloud SQL (Connections - of *gcpcloudsqldb://* type). + """ + Serves DB connection configuration for Google Cloud SQL (Connections of *gcpcloudsqldb://* type). The hook is a "meta" one. It does not perform an actual connection. + It is there to retrieve all the parameters configured in gcpcloudsql:// connection, start/stop Cloud SQL Proxy if needed, dynamically generate Postgres or MySQL connection in the database and return an actual Postgres or MySQL hook. diff --git a/airflow/providers/google/cloud/hooks/compute.py b/airflow/providers/google/cloud/hooks/compute.py index 064551cc40dee..1a95e94c317f4 100644 --- a/airflow/providers/google/cloud/hooks/compute.py +++ b/airflow/providers/google/cloud/hooks/compute.py @@ -107,6 +107,7 @@ def insert_instance_template( ) -> None: """ Creates Instance Template using body specified. + Must be called with keyword arguments rather than positional. :param body: Instance Template representation as an object. @@ -158,9 +159,10 @@ def delete_instance_template( ) -> None: """ Deletes Instance Template. - Deleting an Instance Template is permanent and cannot be undone. It - is not possible to delete templates that are already in use by a managed instance group. - Must be called with keyword arguments rather than positional. + + Deleting an Instance Template is permanent and cannot be undone. It is not + possible to delete templates that are already in use by a managed instance + group. Must be called with keyword arguments rather than positional. :param resource_id: Name of the Compute Engine Instance Template resource. :param request_id: Unique request_id that you might add to achieve @@ -210,6 +212,7 @@ def get_instance_template( ) -> InstanceTemplate: """ Retrieves Instance Template by project_id and resource_id. + Must be called with keyword arguments rather than positional. :param resource_id: Name of the Instance Template. @@ -259,6 +262,7 @@ def insert_instance( ) -> None: """ Creates Instance using body specified. + Must be called with keyword arguments rather than positional. :param body: Instance representation as an object. Should at least include 'name', 'machine_type', @@ -332,6 +336,7 @@ def get_instance( ) -> Instance: """ Retrieves Instance by project_id and resource_id. + Must be called with keyword arguments rather than positional. :param resource_id: Name of the Instance @@ -383,8 +388,8 @@ def delete_instance( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ - Deletes Instance. - Deleting an Instance is permanent and cannot be undone. + Permanently and irrevocably deletes an Instance. + It is not possible to delete Instances that are already in use by a managed instance group. Must be called with keyword arguments rather than positional. @@ -433,6 +438,7 @@ def delete_instance( def start_instance(self, zone: str, resource_id: str, project_id: str) -> None: """ Starts an existing instance defined by project_id, zone and resource_id. + Must be called with keyword arguments rather than positional. :param zone: Google Cloud zone where the instance exists @@ -457,7 +463,8 @@ def start_instance(self, zone: str, resource_id: str, project_id: str) -> None: @GoogleBaseHook.fallback_to_default_project_id def stop_instance(self, zone: str, resource_id: str, project_id: str) -> None: """ - Stops an instance defined by project_id, zone and resource_id + Stops an instance defined by project_id, zone and resource_id. + Must be called with keyword arguments rather than positional. :param zone: Google Cloud zone where the instance exists @@ -483,6 +490,7 @@ def stop_instance(self, zone: str, resource_id: str, project_id: str) -> None: def set_machine_type(self, zone: str, resource_id: str, body: dict, project_id: str) -> None: """ Sets machine type of an instance defined by project_id, zone and resource_id. + Must be called with keyword arguments rather than positional. :param zone: Google Cloud zone where the instance exists. @@ -524,6 +532,7 @@ def insert_instance_group_manager( ) -> None: """ Creates an Instance Group Managers using the body specified. + After the group is created, instances in the group are created using the specified Instance Template. Must be called with keyword arguments rather than positional. @@ -580,6 +589,7 @@ def get_instance_group_manager( ) -> InstanceGroupManager: """ Retrieves Instance Group Manager by project_id, zone and resource_id. + Must be called with keyword arguments rather than positional. :param resource_id: The name of the Managed Instance Group @@ -631,8 +641,8 @@ def delete_instance_group_manager( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ - Deletes Instance Group Managers. - Deleting an Instance Group Manager is permanent and cannot be undone. + Permanently and irrevocably deletes Instance Group Managers. + Must be called with keyword arguments rather than positional. :param resource_id: Name of the Compute Engine Instance Group Managers resource. @@ -687,6 +697,7 @@ def patch_instance_group_manager( ) -> None: """ Patches Instance Group Manager with the specified body. + Must be called with keyword arguments rather than positional. :param zone: Google Cloud zone where the Instance Group Manager exists diff --git a/airflow/providers/google/cloud/hooks/dataflow.py b/airflow/providers/google/cloud/hooks/dataflow.py index 3e0b07fcb28d9..19342cb26b083 100644 --- a/airflow/providers/google/cloud/hooks/dataflow.py +++ b/airflow/providers/google/cloud/hooks/dataflow.py @@ -389,8 +389,7 @@ def _refresh_jobs(self) -> None: def _check_dataflow_job_state(self, job) -> bool: """ - Helper method to check the state of one job in dataflow for this task - if job failed raise exception. + Helper method to check the state of one job in dataflow for this task if job failed raise exception. :return: True if job is done. :raise: Exception diff --git a/airflow/providers/google/cloud/hooks/datafusion.py b/airflow/providers/google/cloud/hooks/datafusion.py index a736d11e9784c..f74958cc16079 100644 --- a/airflow/providers/google/cloud/hooks/datafusion.py +++ b/airflow/providers/google/cloud/hooks/datafusion.py @@ -103,10 +103,7 @@ def wait_for_pipeline_state( failure_states: list[str] | None = None, timeout: int = 5 * 60, ) -> None: - """ - Polls pipeline state and raises an exception if the state is one of - `failure_states` or the operation timed_out. - """ + """Polls pipeline state and raises an exception if the state fails or times out.""" failure_states = failure_states or FAILURE_STATES success_states = success_states or SUCCESS_STATES start_time = monotonic() @@ -190,6 +187,7 @@ def get_conn(self) -> Resource: def restart_instance(self, instance_name: str, location: str, project_id: str) -> Operation: """ Restart a single Data Fusion instance. + At the end of an operation instance is fully restarted. :param instance_name: The name of the instance to restart. diff --git a/airflow/providers/google/cloud/hooks/dataprep.py b/airflow/providers/google/cloud/hooks/dataprep.py index 7e475ea6a7396..c01a48d5ae421 100644 --- a/airflow/providers/google/cloud/hooks/dataprep.py +++ b/airflow/providers/google/cloud/hooks/dataprep.py @@ -105,6 +105,7 @@ def get_jobs_for_job_group(self, job_id: int) -> dict[str, Any]: def get_job_group(self, job_group_id: int, embed: str, include_deleted: bool) -> dict[str, Any]: """ Get the specified job group. + A job group is a job that is executed from a specific node in a flow. :param job_group_id: The ID of the job that will be fetched diff --git a/airflow/providers/google/cloud/hooks/dlp.py b/airflow/providers/google/cloud/hooks/dlp.py index db198de6c5553..47950579894d0 100644 --- a/airflow/providers/google/cloud/hooks/dlp.py +++ b/airflow/providers/google/cloud/hooks/dlp.py @@ -16,8 +16,7 @@ # specific language governing permissions and limitations # under the License. """ -This module contains a CloudDLPHook -which allows you to connect to Google Cloud DLP service. +This module contains a CloudDLPHook which allows you to connect to Google Cloud DLP service. .. spelling:word-list:: @@ -65,6 +64,7 @@ class CloudDLPHook(GoogleBaseHook): """ Hook for Google Cloud Data Loss Prevention (DLP) APIs. + Cloud DLP allows clients to detect the presence of Personally Identifiable Information (PII) and other privacy-sensitive data in user-supplied, unstructured data streams, like text blocks or images. The service also @@ -167,8 +167,7 @@ def create_deidentify_template( metadata: Sequence[tuple[str, str]] = (), ) -> DeidentifyTemplate: """ - Creates a deidentify template for re-using frequently used configuration for - de-identifying content, images, and storage. + Create a deidentify template to reuse frequently-used configurations for content, images, and storage. :param organization_id: (Optional) The organization ID. Required to set this field if parent resource is an organization. @@ -293,8 +292,7 @@ def create_inspect_template( metadata: Sequence[tuple[str, str]] = (), ) -> InspectTemplate: """ - Creates an inspect template for re-using frequently used configuration for - inspecting content, images, and storage. + Create an inspect template to reuse frequently used configurations for content, images, and storage. :param organization_id: (Optional) The organization ID. Required to set this field if parent resource is an organization. @@ -344,8 +342,7 @@ def create_job_trigger( metadata: Sequence[tuple[str, str]] = (), ) -> JobTrigger: """ - Creates a job trigger to run DLP actions such as scanning storage for sensitive - information on a set schedule. + Create a job trigger to run DLP actions such as scanning storage for sensitive info on a set schedule. :param project_id: (Optional) Google Cloud project ID where the DLP Instance exists. If set to None or missing, the default @@ -437,8 +434,7 @@ def deidentify_content( metadata: Sequence[tuple[str, str]] = (), ) -> DeidentifyContentResponse: """ - De-identifies potentially sensitive info from a content item. This method has limits - on input size and output size. + De-identifies potentially sensitive info from a content item; limits input size and output size. :param project_id: (Optional) Google Cloud project ID where the DLP Instance exists. If set to None or missing, the default @@ -531,8 +527,10 @@ def delete_dlp_job( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ - Deletes a long-running DLP job. This method indicates that the client is no longer - interested in the DLP job result. The job will be cancelled if possible. + Deletes a long-running DLP job. + + This method indicates that the client is no longer interested in the DLP job result. + The job will be cancelled if possible. :param dlp_job_id: The ID of the DLP job resource to be cancelled. :param project_id: (Optional) Google Cloud project ID where the @@ -931,8 +929,7 @@ def inspect_content( metadata: Sequence[tuple[str, str]] = (), ) -> InspectContentResponse: """ - Finds potentially sensitive info in content. This method has limits on input size, - processing time, and output size. + Finds potentially sensitive info in content; limits input size, processing time, and output size. :param project_id: (Optional) Google Cloud project ID where the DLP Instance exists. If set to None or missing, the default @@ -1264,8 +1261,7 @@ def redact_image( metadata: Sequence[tuple[str, str]] = (), ) -> RedactImageResponse: """ - Redacts potentially sensitive info from an image. This method has limits on - input size, processing time, and output size. + Redacts potentially sensitive info from an image; limits input size, processing time, and output size. :param project_id: (Optional) Google Cloud project ID where the DLP Instance exists. If set to None or missing, the default diff --git a/airflow/providers/google/cloud/hooks/gcs.py b/airflow/providers/google/cloud/hooks/gcs.py index 42a27e80f898c..1b8256893f0b7 100644 --- a/airflow/providers/google/cloud/hooks/gcs.py +++ b/airflow/providers/google/cloud/hooks/gcs.py @@ -143,10 +143,7 @@ def _inner_wrapper(self, *args, **kwargs) -> RT: class GCSHook(GoogleBaseHook): - """ - Interact with Google Cloud Storage. This hook uses the Google Cloud - connection. - """ + """Use the Google Cloud connection to interact with Google Cloud Storage.""" _conn: storage.Client | None = None @@ -232,9 +229,7 @@ def rewrite( destination_object: str | None = None, ) -> None: """ - Has the same functionality as copy, except that will work on files - over 5 TB, as well as when copying between locations and/or storage - classes. + Similar to copy; supports files over 5 TB, and copying between locations and/or storage classes. destination_object can be omitted, in which case source_object is used. @@ -430,8 +425,7 @@ def provide_file_and_upload( object_url: str | None = None, ) -> Generator[IO[bytes], None, None]: """ - Creates temporary file, returns a file handle and uploads the files content - on close. + Creates temporary file, returns a file handle and uploads the files content on close. You can use this method by passing the bucket_name and object_name parameters or just object_url parameter. @@ -483,7 +477,9 @@ def upload( """ def _call_with_retry(f: Callable[[], None]) -> None: - """Helper functions to upload a file or a string with a retry mechanism and exponential back-off. + """ + Helper functions to upload a file or a string with a retry mechanism and exponential back-off. + :param f: Callable that should be retried. """ num_file_attempts = 0 @@ -834,6 +830,7 @@ def _list_blobs_with_match_glob( ) -> Any: """ List blobs when match_glob param is given. + This method is a patched version of google.cloud.storage Client.list_blobs(). It is used as a temporary workaround to support "match_glob" param, as it isn't officially supported by GCS Python client. @@ -879,8 +876,7 @@ def list_by_timespan( match_glob: str | None = None, ) -> List[str]: """ - List all objects from the bucket with the give string prefix in name that were - updated in the time between ``timespan_start`` and ``timespan_end``. + List all objects from the bucket with the given string prefix that were updated in the time range. :param bucket_name: bucket name :param timespan_start: will return objects that were updated at or after this datetime (UTC) @@ -1002,8 +998,10 @@ def create_bucket( labels: dict | None = None, ) -> str: """ - Creates a new bucket. Google Cloud Storage uses a flat namespace, so - you can't create a bucket with a name that is already in use. + Creates a new bucket. + + Google Cloud Storage uses a flat namespace, so you can't + create a bucket with a name that is already in use. .. seealso:: For more information, see Bucket Naming Guidelines: @@ -1314,10 +1312,7 @@ def _prepare_sync_plan( def gcs_object_is_directory(bucket: str) -> bool: - """ - Return True if given Google Cloud Storage URL (gs:///) - is a directory or an empty bucket. Otherwise return False. - """ + """Return True if given Google Cloud Storage URL (gs:///) is a directory or empty bucket.""" _, blob = _parse_gcs_url(bucket) return len(blob) == 0 or blob.endswith("/") @@ -1355,8 +1350,9 @@ def parse_json_from_gcs(gcp_conn_id: str, file_uri: str) -> Any: def _parse_gcs_url(gsurl: str) -> tuple[str, str]: """ - Given a Google Cloud Storage URL (gs:///), returns a - tuple containing the corresponding bucket and blob. + Given a Google Cloud Storage URL, return a tuple containing the corresponding bucket and blob. + + Expected url format: gs:/// """ parsed_url = urlsplit(gsurl) if not parsed_url.netloc: diff --git a/airflow/providers/google/cloud/hooks/gdm.py b/airflow/providers/google/cloud/hooks/gdm.py index 7d228c8c95efc..203191d1644d5 100644 --- a/airflow/providers/google/cloud/hooks/gdm.py +++ b/airflow/providers/google/cloud/hooks/gdm.py @@ -28,6 +28,7 @@ class GoogleDeploymentManagerHook(GoogleBaseHook): """ Interact with Google Cloud Deployment Manager using the Google Cloud connection. + This allows for scheduled and programmatic inspection and deletion of resources managed by GDM. """ diff --git a/airflow/providers/google/cloud/hooks/life_sciences.py b/airflow/providers/google/cloud/hooks/life_sciences.py index 835837ef5fb81..e3cd01c6c3df2 100644 --- a/airflow/providers/google/cloud/hooks/life_sciences.py +++ b/airflow/providers/google/cloud/hooks/life_sciences.py @@ -122,8 +122,7 @@ def _location_path(self, project_id: str, location: str) -> str: def _wait_for_operation_to_complete(self, operation_name: str) -> None: """ - Waits for the named operation to complete - checks status of the - asynchronous call. + Waits for the named operation to complete - checks status of the asynchronous call. :param operation_name: The name of the operation. :return: The response returned by the operation. diff --git a/airflow/providers/google/cloud/hooks/looker.py b/airflow/providers/google/cloud/hooks/looker.py index 42477d2fbbba2..d1179d707caee 100644 --- a/airflow/providers/google/cloud/hooks/looker.py +++ b/airflow/providers/google/cloud/hooks/looker.py @@ -209,8 +209,9 @@ def __init__( def read_config(self): """ - Overrides the default logic of getting connection settings. Fetches - the connection settings from Airflow's connection object. + Fetches the connection settings from Airflow's connection object. + + Overrides the default logic of getting connection settings. """ config = {} diff --git a/airflow/providers/google/cloud/hooks/natural_language.py b/airflow/providers/google/cloud/hooks/natural_language.py index ee6f25965cf6c..673df34b23280 100644 --- a/airflow/providers/google/cloud/hooks/natural_language.py +++ b/airflow/providers/google/cloud/hooks/natural_language.py @@ -90,8 +90,9 @@ def analyze_entities( metadata: Sequence[tuple[str, str]] = (), ) -> AnalyzeEntitiesResponse: """ - Finds named entities in the text along with entity types, - salience, mentions for each entity, and other properties. + Finds named entities in the text along with various properties. + + Examples properties: entity types, salience, mentions for each entity, and others. :param document: Input document. If a dict is provided, it must be of the same form as the protobuf message Document @@ -120,8 +121,7 @@ def analyze_entity_sentiment( metadata: Sequence[tuple[str, str]] = (), ) -> AnalyzeEntitySentimentResponse: """ - Finds entities, similar to AnalyzeEntities in the text and analyzes sentiment associated with each - entity and its mentions. + Similar to AnalyzeEntities, also analyzes sentiment associated with each entity and its mentions. :param document: Input document. If a dict is provided, it must be of the same form as the protobuf message Document @@ -179,7 +179,9 @@ def analyze_syntax( metadata: Sequence[tuple[str, str]] = (), ) -> AnalyzeSyntaxResponse: """ - Analyzes the syntax of the text and provides sentence boundaries and tokenization along with part + Analyzes the syntax of the text. + + Provides sentence boundaries and tokenization along with part of speech tags, dependency trees, and other properties. :param document: Input document. @@ -210,8 +212,7 @@ def annotate_text( metadata: Sequence[tuple[str, str]] = (), ) -> AnnotateTextResponse: """ - A convenience method that provides all the features that analyzeSentiment, - analyzeEntities, and analyzeSyntax provide in one call. + Provide all features that analyzeSentiment, analyzeEntities, and analyzeSyntax provide in one call. :param document: Input document. If a dict is provided, it must be of the same form as the protobuf message Document diff --git a/airflow/providers/google/cloud/hooks/os_login.py b/airflow/providers/google/cloud/hooks/os_login.py index c82772cc821b5..9a355d74d9ea7 100644 --- a/airflow/providers/google/cloud/hooks/os_login.py +++ b/airflow/providers/google/cloud/hooks/os_login.py @@ -76,9 +76,9 @@ def import_ssh_public_key( metadata: Sequence[tuple[str, str]] = (), ) -> ImportSshPublicKeyResponse: """ - Adds an SSH public key and returns the profile information. Default POSIX - account information is set when no username and UID exist as part of the - login profile. + Adds an SSH public key and returns the profile information. + + Default POSIX account information is set when no username and UID exist as part of the login profile. :param user: The unique ID for the user :param ssh_public_key: The SSH public key and expiration time. diff --git a/airflow/providers/google/cloud/hooks/pubsub.py b/airflow/providers/google/cloud/hooks/pubsub.py index 0e33f22ebf59c..6e867c2bdb1ff 100644 --- a/airflow/providers/google/cloud/hooks/pubsub.py +++ b/airflow/providers/google/cloud/hooks/pubsub.py @@ -587,7 +587,8 @@ def __init__(self, project_id: str | None = None, **kwargs: Any): async def _get_subscriber_client(self) -> SubscriberAsyncClient: """ - Returns async connection to the Google PubSub + Returns async connection to the Google PubSub. + :return: Google Pub/Sub asynchronous client. """ if not self._client: diff --git a/airflow/providers/google/cloud/hooks/spanner.py b/airflow/providers/google/cloud/hooks/spanner.py index d83cfc598ad8d..00fe9a880456e 100644 --- a/airflow/providers/google/cloud/hooks/spanner.py +++ b/airflow/providers/google/cloud/hooks/spanner.py @@ -215,8 +215,7 @@ def get_database( project_id: str, ) -> Database | None: """ - Retrieves a database in Cloud Spanner. If the database does not exist - in the specified instance, it returns None. + Retrieves a database in Cloud Spanner; return None if the database does not exist in the instance. :param instance_id: The ID of the Cloud Spanner instance. :param database_id: The ID of the database in Cloud Spanner. diff --git a/airflow/providers/google/cloud/hooks/stackdriver.py b/airflow/providers/google/cloud/hooks/stackdriver.py index 517e211edd0df..28e0b8830357a 100644 --- a/airflow/providers/google/cloud/hooks/stackdriver.py +++ b/airflow/providers/google/cloud/hooks/stackdriver.py @@ -77,10 +77,10 @@ def list_alert_policies( metadata: Sequence[tuple[str, str]] = (), ) -> Any: """ - Fetches all the Alert Policies identified by the filter passed as - filter parameter. The desired return type can be specified by the - format parameter, the supported formats are "dict", "json" and None - which returns python dictionary, stringified JSON and protobuf + Fetches all the Alert Policies identified by the filter passed as filter parameter. + + The desired return type can be specified by the format parameter, the supported formats + are "dict", "json" and None which returns python dictionary, stringified JSON and protobuf respectively. :param format_: (Optional) Desired output format of the result. The @@ -158,8 +158,9 @@ def enable_alert_policies( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ - Enables one or more disabled alerting policies identified by filter - parameter. Inoperative in case the policy is already enabled. + Enables one or more disabled alerting policies identified by filter parameter. + + Inoperative in case the policy is already enabled. :param project_id: The project in which alert needs to be enabled. :param filter_: If provided, this field specifies the criteria that @@ -191,8 +192,9 @@ def disable_alert_policies( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ - Disables one or more enabled alerting policies identified by filter - parameter. Inoperative in case the policy is already disabled. + Disables one or more enabled alerting policies identified by filter parameter. + + Inoperative in case the policy is already disabled. :param project_id: The project in which alert needs to be disabled. :param filter_: If provided, this field specifies the criteria that @@ -224,8 +226,7 @@ def upsert_alert( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ - Creates a new alert or updates an existing policy identified - the name field in the alerts parameter. + Creates a new alert or updates an existing policy identified the name field in the alerts parameter. :param project_id: The project in which alert needs to be created/updated. :param alerts: A JSON string or file that specifies all the alerts that needs @@ -352,10 +353,10 @@ def list_notification_channels( metadata: Sequence[tuple[str, str]] = (), ) -> Any: """ - Fetches all the Notification Channels identified by the filter passed as - filter parameter. The desired return type can be specified by the - format parameter, the supported formats are "dict", "json" and None - which returns python dictionary, stringified JSON and protobuf + Fetches all the Notification Channels identified by the filter passed as filter parameter. + + The desired return type can be specified by the format parameter, the supported formats are + "dict", "json" and None which returns python dictionary, stringified JSON and protobuf respectively. :param format_: (Optional) Desired output format of the result. The @@ -435,8 +436,9 @@ def enable_notification_channels( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ - Enables one or more disabled alerting policies identified by filter - parameter. Inoperative in case the policy is already enabled. + Enables one or more disabled alerting policies identified by filter parameter. + + Inoperative in case the policy is already enabled. :param project_id: The project in which notification channels needs to be enabled. :param filter_: If provided, this field specifies the criteria that @@ -468,8 +470,9 @@ def disable_notification_channels( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ - Disables one or more enabled notification channels identified by filter - parameter. Inoperative in case the policy is already disabled. + Disables one or more enabled notification channels identified by filter parameter. + + Inoperative in case the policy is already disabled. :param project_id: The project in which notification channels needs to be enabled. :param filter_: If provided, this field specifies the criteria that @@ -501,8 +504,9 @@ def upsert_channel( metadata: Sequence[tuple[str, str]] = (), ) -> dict: """ - Creates a new notification or updates an existing notification channel - identified the name field in the alerts parameter. + Create a new notification or updates an existing notification channel. + + Channel is identified by the name field in the alerts parameter. :param channels: A JSON string or file that specifies all the alerts that needs to be either created or updated. For more details, see diff --git a/airflow/providers/google/cloud/hooks/tasks.py b/airflow/providers/google/cloud/hooks/tasks.py index 1301a3406e76a..4807d0fde77cc 100644 --- a/airflow/providers/google/cloud/hooks/tasks.py +++ b/airflow/providers/google/cloud/hooks/tasks.py @@ -15,11 +15,8 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -""" -This module contains a CloudTasksHook -which allows you to connect to Google Cloud Tasks service, -performing actions to queues or tasks. -""" +"""This module contains a CloudTasksHook which allows you to connect to Google Cloud Tasks service.""" + from __future__ import annotations from typing import Sequence @@ -37,8 +34,9 @@ class CloudTasksHook(GoogleBaseHook): """ - Hook for Google Cloud Tasks APIs. Cloud Tasks allows developers to manage - the execution of background work in their applications. + Hook for Google Cloud Tasks APIs. + + Cloud Tasks allows developers to manage the execution of background work in their applications. All the methods in the hook where project_id is used must be called with keyword arguments rather than positional. diff --git a/airflow/providers/google/cloud/hooks/vertex_ai/custom_job.py b/airflow/providers/google/cloud/hooks/vertex_ai/custom_job.py index 3686005713678..b5f43d643f1e7 100644 --- a/airflow/providers/google/cloud/hooks/vertex_ai/custom_job.py +++ b/airflow/providers/google/cloud/hooks/vertex_ai/custom_job.py @@ -357,7 +357,9 @@ def cancel_pipeline_job( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ - Cancels a PipelineJob. Starts asynchronous cancellation on the PipelineJob. The server makes a best + Cancels a PipelineJob. + + Starts asynchronous cancellation on the PipelineJob. The server makes a best effort to cancel the pipeline, but success is not guaranteed. Clients can use [PipelineService.GetPipelineJob][google.cloud.aiplatform.v1.PipelineService.GetPipelineJob] or other methods to check whether the cancellation succeeded or whether the pipeline completed despite @@ -396,7 +398,9 @@ def cancel_training_pipeline( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ - Cancels a TrainingPipeline. Starts asynchronous cancellation on the TrainingPipeline. The server makes + Cancels a TrainingPipeline. + + Starts asynchronous cancellation on the TrainingPipeline. The server makes a best effort to cancel the pipeline, but success is not guaranteed. Clients can use [PipelineService.GetTrainingPipeline][google.cloud.aiplatform.v1.PipelineService.GetTrainingPipeline] or other methods to check whether the cancellation succeeded or whether the pipeline completed despite @@ -435,7 +439,9 @@ def cancel_custom_job( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ - Cancels a CustomJob. Starts asynchronous cancellation on the CustomJob. The server makes a best effort + Cancels a CustomJob. + + Starts asynchronous cancellation on the CustomJob. The server makes a best effort to cancel the job, but success is not guaranteed. Clients can use [JobService.GetCustomJob][google.cloud.aiplatform.v1.JobService.GetCustomJob] or other methods to check whether the cancellation succeeded or whether the job completed despite cancellation. On diff --git a/airflow/providers/google/cloud/hooks/vertex_ai/endpoint_service.py b/airflow/providers/google/cloud/hooks/vertex_ai/endpoint_service.py index 2750f70241ad7..3c1fb9b3d019e 100644 --- a/airflow/providers/google/cloud/hooks/vertex_ai/endpoint_service.py +++ b/airflow/providers/google/cloud/hooks/vertex_ai/endpoint_service.py @@ -316,8 +316,7 @@ def undeploy_model( metadata: Sequence[tuple[str, str]] = (), ) -> Operation: """ - Undeploys a Model from an Endpoint, removing a DeployedModel from it, and freeing all resources it's - using. + Undeploys a Model from an Endpoint, removing a DeployedModel from it, and freeing all used resources. :param project_id: Required. The ID of the Google Cloud project that the service belongs to. :param region: Required. The ID of the Google Cloud region that the service belongs to. diff --git a/airflow/providers/google/cloud/hooks/vision.py b/airflow/providers/google/cloud/hooks/vision.py index e7cc242e4b9b0..a6e504ab4fcb6 100644 --- a/airflow/providers/google/cloud/hooks/vision.py +++ b/airflow/providers/google/cloud/hooks/vision.py @@ -64,20 +64,16 @@ def get_entity_with_name( self, entity: Any, entity_id: str | None, location: str | None, project_id: str ) -> Any: """ - Check if entity has the `name` attribute set: - * If so, no action is taken. + Check if entity has the `name` attribute set. + * If so, no action is taken. * If not, and the name can be constructed from other parameters provided, it is created and filled in the entity. - * If both the entity's 'name' attribute is set and the name can be constructed from other parameters provided: - * If they are the same - no action is taken - * if they are different - an exception is thrown. - :param entity: Entity :param entity_id: Entity id :param location: Location @@ -178,6 +174,8 @@ def create_product_set( metadata: Sequence[tuple[str, str]] = (), ) -> str: """ + Create product set. + For the documentation see: :class:`~airflow.providers.google.cloud.operators.vision.CloudVisionCreateProductSetOperator`. """ @@ -213,6 +211,8 @@ def get_product_set( metadata: Sequence[tuple[str, str]] = (), ) -> dict: """ + Get product set. + For the documentation see: :class:`~airflow.providers.google.cloud.operators.vision.CloudVisionGetProductSetOperator`. """ @@ -237,6 +237,8 @@ def update_product_set( metadata: Sequence[tuple[str, str]] = (), ) -> dict: """ + Update product set. + For the documentation see: :class:`~airflow.providers.google.cloud.operators.vision.CloudVisionUpdateProductSetOperator`. """ @@ -270,6 +272,8 @@ def delete_product_set( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ + Delete product set. + For the documentation see: :class:`~airflow.providers.google.cloud.operators.vision.CloudVisionDeleteProductSetOperator`. """ @@ -291,6 +295,8 @@ def create_product( metadata: Sequence[tuple[str, str]] = (), ): """ + Create product. + For the documentation see: :class:`~airflow.providers.google.cloud.operators.vision.CloudVisionCreateProductOperator`. """ @@ -329,6 +335,8 @@ def get_product( metadata: Sequence[tuple[str, str]] = (), ): """ + Get product. + For the documentation see: :class:`~airflow.providers.google.cloud.operators.vision.CloudVisionGetProductOperator`. """ @@ -353,6 +361,8 @@ def update_product( metadata: Sequence[tuple[str, str]] = (), ): """ + Update product. + For the documentation see: :class:`~airflow.providers.google.cloud.operators.vision.CloudVisionUpdateProductOperator`. """ @@ -384,6 +394,8 @@ def delete_product( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ + Delete product. + For the documentation see: :class:`~airflow.providers.google.cloud.operators.vision.CloudVisionDeleteProductOperator`. """ @@ -406,6 +418,8 @@ def create_reference_image( metadata: Sequence[tuple[str, str]] = (), ) -> str: """ + Create reference image. + For the documentation see: :py:class:`~airflow.providers.google.cloud.operators.vision.CloudVisionCreateReferenceImageOperator`. """ @@ -448,6 +462,8 @@ def delete_reference_image( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ + Delete reference image. + For the documentation see: :py:class:`~airflow.providers.google.cloud.operators.vision.CloudVisionDeleteReferenceImageOperator`. """ @@ -478,6 +494,8 @@ def add_product_to_product_set( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ + Add product to product set. + For the documentation see: :py:class:`~airflow.providers.google.cloud.operators.vision.CloudVisionAddProductToProductSetOperator`. """ @@ -506,6 +524,8 @@ def remove_product_from_product_set( metadata: Sequence[tuple[str, str]] = (), ) -> None: """ + Remove product from product set. + For the documentation see: :py:class:`~airflow.providers.google.cloud.operators.vision.CloudVisionRemoveProductFromProductSetOperator`. """ @@ -529,6 +549,8 @@ def annotate_image( timeout: float | None = None, ) -> dict: """ + Annotate image. + For the documentation see: :py:class:`~airflow.providers.google.cloud.operators.vision.CloudVisionImageAnnotateOperator`. """ @@ -550,6 +572,8 @@ def batch_annotate_images( timeout: float | None = None, ) -> dict: """ + Batch annotate images. + For the documentation see: :py:class:`~airflow.providers.google.cloud.operators.vision.CloudVisionImageAnnotateOperator`. """ @@ -574,6 +598,8 @@ def text_detection( additional_properties: dict | None = None, ) -> dict: """ + Text detection. + For the documentation see: :py:class:`~airflow.providers.google.cloud.operators.vision.CloudVisionDetectTextOperator`. """ @@ -604,6 +630,8 @@ def document_text_detection( additional_properties: dict | None = None, ) -> dict: """ + Document text detection. + For the documentation see: :py:class:`~airflow.providers.google.cloud.operators.vision.CloudVisionTextDetectOperator`. """ @@ -634,6 +662,8 @@ def label_detection( additional_properties: dict | None = None, ) -> dict: """ + Label detection. + For the documentation see: :py:class:`~airflow.providers.google.cloud.operators.vision.CloudVisionDetectImageLabelsOperator`. """ @@ -664,6 +694,8 @@ def safe_search_detection( additional_properties: dict | None = None, ) -> dict: """ + Safe search detection. + For the documentation see: :py:class:`~airflow.providers.google.cloud.operators.vision.CloudVisionDetectImageSafeSearchOperator`. """ diff --git a/airflow/providers/google/cloud/hooks/workflows.py b/airflow/providers/google/cloud/hooks/workflows.py index 6e1f6ed1209eb..6e3942e3b9659 100644 --- a/airflow/providers/google/cloud/hooks/workflows.py +++ b/airflow/providers/google/cloud/hooks/workflows.py @@ -67,10 +67,11 @@ def create_workflow( metadata: Sequence[tuple[str, str]] = (), ) -> Operation: """ - Creates a new workflow. If a workflow with the specified name - already exists in the specified project and location, the long - running operation will return - [ALREADY_EXISTS][google.rpc.Code.ALREADY_EXISTS] error. + Creates a new workflow. + + If a workflow with the specified name already exists in the + specified project and location, the long running operation will + return [ALREADY_EXISTS][google.rpc.Code.ALREADY_EXISTS] error. :param workflow: Required. Workflow to be created. :param workflow_id: Required. The ID of the workflow to be created. @@ -129,6 +130,7 @@ def update_workflow( ) -> Operation: """ Updates an existing workflow. + Running this method has no impact on already running executions of the workflow. A new revision of the workflow may be created as a result of a successful @@ -164,9 +166,7 @@ def delete_workflow( metadata: Sequence[tuple[str, str]] = (), ) -> Operation: """ - Deletes a workflow with the specified name. - This method also cancels and deletes all running - executions of the workflow. + Delete a workflow with the specified name and all running executions of the workflow. :param workflow_id: Required. The ID of the workflow to be created. :param project_id: Required. The ID of the Google Cloud project the cluster belongs to. @@ -194,8 +194,7 @@ def list_workflows( metadata: Sequence[tuple[str, str]] = (), ) -> ListWorkflowsPager: """ - Lists Workflows in a given project and location. - The default order is not specified. + Lists Workflows in a given project and location; the default order is not specified. :param filter_: Filter to restrict results to specific workflows. :param order_by: Comma-separated list of fields that @@ -233,8 +232,7 @@ def create_execution( metadata: Sequence[tuple[str, str]] = (), ) -> Execution: """ - Creates a new execution using the latest revision of - the given workflow. + Creates a new execution using the latest revision of the given workflow. :param execution: Required. Input parameters of the execution represented as a dictionary. :param workflow_id: Required. The ID of the workflow. @@ -328,11 +326,10 @@ def list_executions( metadata: Sequence[tuple[str, str]] = (), ) -> ListExecutionsPager: """ - Returns a list of executions which belong to the - workflow with the given name. The method returns - executions of all workflow revisions. Returned - executions are ordered by their start time (newest - first). + Returns a list of executions which belong to the workflow with the given name. + + The method returns executions of all workflow revisions. Returned + executions are ordered by their start time (newest first). :param workflow_id: Required. The ID of the workflow to be created. :param project_id: Required. The ID of the Google Cloud project the cluster belongs to. diff --git a/airflow/providers/google/cloud/log/gcs_task_handler.py b/airflow/providers/google/cloud/log/gcs_task_handler.py index 0c3f8a27c73c2..121a668857b29 100644 --- a/airflow/providers/google/cloud/log/gcs_task_handler.py +++ b/airflow/providers/google/cloud/log/gcs_task_handler.py @@ -59,10 +59,10 @@ def get_default_delete_local_copy(): class GCSTaskHandler(FileTaskHandler, LoggingMixin): """ - GCSTaskHandler is a python log handler that handles and reads - task instance logs. It extends airflow FileTaskHandler and - uploads to and reads from GCS remote storage. Upon log reading - failure, it reads from host machine's local disk. + GCSTaskHandler is a python log handler that handles and reads task instance logs. + + It extends airflow FileTaskHandler and uploads to and reads from GCS remote + storage. Upon log reading failure, it reads from host machine's local disk. :param base_log_folder: Base log folder to place logs. :param gcs_log_folder: Path to a remote location where logs will be saved. It must have the prefix @@ -209,6 +209,7 @@ def _read_remote_logs(self, ti, try_number, metadata=None) -> tuple[list[str], l def _read(self, ti, try_number, metadata=None): """ Read logs of given task instance and try_number from GCS. + If failed, read the log from task instance host machine. todo: when min airflow version >= 2.6, remove this method @@ -231,8 +232,7 @@ def _read(self, ti, try_number, metadata=None): def gcs_write(self, log, remote_log_location) -> bool: """ - Writes the log to the remote_log_location and return `True` when done. Fails silently - and return `False` if no log was created. + Write the log to the remote location and return `True`; fail silently and return `False` on error. :param log: the log to write to the remote_log_location :param remote_log_location: the log's location in remote storage diff --git a/airflow/providers/google/cloud/operators/automl.py b/airflow/providers/google/cloud/operators/automl.py index 0503e6bffefef..e590dc3d06207 100644 --- a/airflow/providers/google/cloud/operators/automl.py +++ b/airflow/providers/google/cloud/operators/automl.py @@ -895,8 +895,9 @@ def execute(self, context: Context): class AutoMLDeployModelOperator(GoogleCloudBaseOperator): """ - Deploys a model. If a model is already deployed, deploying it with the same parameters - has no effect. Deploying with different parameters (as e.g. changing node_number) will + Deploys a model; if a model is already deployed, deploying it with the same parameters has no effect. + + Deploying with different parameters (as e.g. changing node_number) will reset the deployment state without pausing the model_id's availability. Only applicable for Text Classification, Image Object Detection and Tables; all other diff --git a/airflow/providers/google/cloud/operators/bigquery.py b/airflow/providers/google/cloud/operators/bigquery.py index 98929b6e6dec6..5ac3e8671e410 100644 --- a/airflow/providers/google/cloud/operators/bigquery.py +++ b/airflow/providers/google/cloud/operators/bigquery.py @@ -401,8 +401,7 @@ def execute_complete(self, context: Context, event: dict[str, Any]) -> None: class BigQueryIntervalCheckOperator(_BigQueryDbHookMixin, SQLIntervalCheckOperator): """ - Checks that the values of metrics given as SQL expressions are within - a certain tolerance of the ones from days_back before. + Check that the values of metrics given as SQL expressions are within a tolerance of the older ones. This method constructs a query like so :: @@ -544,9 +543,9 @@ def execute_complete(self, context: Context, event: dict[str, Any]) -> None: class BigQueryColumnCheckOperator(_BigQueryDbHookMixin, SQLColumnCheckOperator): """ - BigQueryColumnCheckOperator subclasses the SQLColumnCheckOperator - in order to provide a job id for OpenLineage to parse. See base class - docstring for usage. + Subclasses the SQLColumnCheckOperator in order to provide a job id for OpenLineage to parse. + + See base class docstring for usage. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -667,9 +666,9 @@ def execute(self, context=None): class BigQueryTableCheckOperator(_BigQueryDbHookMixin, SQLTableCheckOperator): """ - BigQueryTableCheckOperator subclasses the SQLTableCheckOperator - in order to provide a job id for OpenLineage to parse. See base class - for usage. + Subclasses the SQLTableCheckOperator in order to provide a job id for OpenLineage to parse. + + See base class for usage. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -769,8 +768,9 @@ def execute(self, context=None): class BigQueryGetDataOperator(GoogleCloudBaseOperator): """ - Fetches the data from a BigQuery table (alternatively fetch data for selected columns) and returns data - in either of the following two formats, based on "as_dict" value: + Fetches the data from a BigQuery table (alternatively fetch data for selected columns) and returns data. + + Data is returned in either of the following two formats, based on "as_dict" value: 1. False (Default) - A Python list of lists, with the number of nested lists equal to the number of rows fetched. Each nested list represents a row, where the elements within it correspond to the column values for that particular row. diff --git a/airflow/providers/google/cloud/operators/bigquery_dts.py b/airflow/providers/google/cloud/operators/bigquery_dts.py index d9e013afa68c3..2d7e4fc02b271 100644 --- a/airflow/providers/google/cloud/operators/bigquery_dts.py +++ b/airflow/providers/google/cloud/operators/bigquery_dts.py @@ -218,10 +218,10 @@ def execute(self, context: Context) -> None: class BigQueryDataTransferServiceStartTransferRunsOperator(GoogleCloudBaseOperator): """ - Start manual transfer runs to be executed now with schedule_time equal - to current time. The transfer runs can be created for a time range where - the run_time is between start_time (inclusive) and end_time - (exclusive), or for a specific run_time. + Start manual transfer runs to be executed now with schedule_time equal to current time. + + The transfer runs can be created for a time range where the run_time is between + start_time (inclusive) and end_time (exclusive), or for a specific run_time. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/operators/bigtable.py b/airflow/providers/google/cloud/operators/bigtable.py index 66d3e46a41dc4..c6af84ddd1a49 100644 --- a/airflow/providers/google/cloud/operators/bigtable.py +++ b/airflow/providers/google/cloud/operators/bigtable.py @@ -52,9 +52,10 @@ def _validate_inputs(self): class BigtableCreateInstanceOperator(GoogleCloudBaseOperator, BigtableValidationMixin): """ Creates a new Cloud Bigtable instance. + If the Cloud Bigtable instance with the given ID exists, the operator does not - compare its configuration - and immediately succeeds. No changes are made to the existing instance. + compare its configuration and immediately succeeds. No changes are made to the + existing instance. For more details about instance creation have a look at the reference: https://googleapis.github.io/google-cloud-python/latest/bigtable/instance.html#google.cloud.bigtable.instance.Instance.create diff --git a/airflow/providers/google/cloud/operators/cloud_base.py b/airflow/providers/google/cloud/operators/cloud_base.py index fc7845e1d8874..bd1194e6ece6d 100644 --- a/airflow/providers/google/cloud/operators/cloud_base.py +++ b/airflow/providers/google/cloud/operators/cloud_base.py @@ -24,14 +24,12 @@ class GoogleCloudBaseOperator(BaseOperator): - """ - Abstract base class that takes care of common specifics of the operators built - on top of Google API client libraries. - """ + """Abstract base class for operators using Google API client libraries.""" def __deepcopy__(self, memo): """ Updating the memo to fix the non-copyable global constant. + This constant can be specified in operator parameters as a retry configuration to indicate a default. See https://github.com/apache/airflow/issues/28751 for details. """ diff --git a/airflow/providers/google/cloud/operators/cloud_build.py b/airflow/providers/google/cloud/operators/cloud_build.py index 4242f561c1003..293d5ddf6f756 100644 --- a/airflow/providers/google/cloud/operators/cloud_build.py +++ b/airflow/providers/google/cloud/operators/cloud_build.py @@ -746,8 +746,7 @@ def execute(self, context: Context): class CloudBuildRetryBuildOperator(GoogleCloudBaseOperator): """ - Creates a new build based on the specified build. This method creates a new build - using the original build request, which may or may not result in an identical build. + Creates a new build using the original build request, which may or may not result in an identical build. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -998,6 +997,7 @@ def execute(self, context: Context): class BuildProcessor: """ Processes build configurations to add additional functionality to support the use of operators. + The following improvements are made: * It is required to provide the source and only one type can be given, * It is possible to provide the source as the URL address instead dict. diff --git a/airflow/providers/google/cloud/operators/cloud_memorystore.py b/airflow/providers/google/cloud/operators/cloud_memorystore.py index 250514388fad9..4c27b8982f70a 100644 --- a/airflow/providers/google/cloud/operators/cloud_memorystore.py +++ b/airflow/providers/google/cloud/operators/cloud_memorystore.py @@ -325,8 +325,9 @@ def execute(self, context: Context) -> None: class CloudMemorystoreFailoverInstanceOperator(GoogleCloudBaseOperator): """ - Initiates a failover of the primary node to current replica node for a specific STANDARD tier Cloud - Memorystore for Redis instance. + Initiate a failover of the primary node for a specific STANDARD tier Cloud Memorystore for Redis instance. + + Uses the current replica node. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -886,8 +887,7 @@ def execute(self, context: Context) -> None: class CloudMemorystoreCreateInstanceAndImportOperator(GoogleCloudBaseOperator): """ - Creates a Redis instance based on the specified tier and memory size and import a Redis RDB snapshot file - from Cloud Storage into a this instance. + Create a Redis instance and import a Redis RDB snapshot file from Cloud Storage into this instance. By default, the instance is accessible from the project's `default network `__. @@ -1007,8 +1007,9 @@ def execute(self, context: Context) -> None: class CloudMemorystoreExportAndDeleteInstanceOperator(GoogleCloudBaseOperator): """ - Export Redis instance data into a Redis RDB format file in Cloud Storage. In next step, deletes a this - instance. + Export Redis instance data into a Redis RDB format file in Cloud Storage. + + In next step, deletes this instance. Redis will continue serving during this operation. @@ -1429,8 +1430,7 @@ def execute(self, context: Context): class CloudMemorystoreMemcachedListInstancesOperator(GoogleCloudBaseOperator): """ - Lists all Memcached instances owned by a project in either the specified location (region) or all - locations. + List all Memcached instances owned by a project in either the specified location/region or all locations. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -1617,9 +1617,10 @@ def execute(self, context: Context): class CloudMemorystoreMemcachedUpdateParametersOperator(GoogleCloudBaseOperator): """ - Updates the defined Memcached Parameters for an existing Instance. This method only stages the - parameters, it must be followed by apply_parameters to apply the parameters to nodes of - the Memcached Instance. + Updates the defined Memcached Parameters for an existing Instance. + + This method only stages the parameters, it must be followed by apply_parameters + to apply the parameters to nodes of the Memcached Instance. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/operators/cloud_sql.py b/airflow/providers/google/cloud/operators/cloud_sql.py index 5c77cbd86c948..868268175e6ea 100644 --- a/airflow/providers/google/cloud/operators/cloud_sql.py +++ b/airflow/providers/google/cloud/operators/cloud_sql.py @@ -1026,8 +1026,8 @@ def execute(self, context: Context) -> None: def execute_complete(self, context, event=None) -> None: """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ if event["status"] == "success": self.log.info("Operation %s completed successfully", event["operation_name"]) diff --git a/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py b/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py index dc7bfc65c5b86..8ae31e33f0f95 100644 --- a/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py +++ b/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py @@ -111,8 +111,7 @@ def _reformat_schedule(self) -> None: def process_body(self) -> dict: """ - Injects AWS credentials into body if needed and - reformats schedule information. + Injects AWS credentials into body if needed and reformats schedule information. :return: Preprocessed body """ @@ -162,9 +161,10 @@ def _restrict_aws_credentials(self) -> None: def validate_body(self) -> None: """ - Validates the body. Checks if body specifies `transferSpec` - if yes, then check if AWS credentials are passed correctly and - no more than 1 data source was selected. + Validates the body. + + Checks if body specifies `transferSpec` if yes, then check if AWS credentials + are passed correctly and no more than 1 data source was selected. :raises: AirflowException """ @@ -360,10 +360,11 @@ def execute(self, context: Context) -> dict: class CloudDataTransferServiceDeleteJobOperator(GoogleCloudBaseOperator): """ - Delete a transfer job. This is a soft delete. After a transfer job is - deleted, the job and all the transfer executions are subject to garbage - collection. Transfer jobs become eligible for garbage collection - 30 days after soft delete. + Delete a transfer job. + + This is a soft delete. After a transfer job is deleted, the job and all the transfer + executions are subject to garbage collection. Transfer jobs become eligible for garbage + collection 30 days after soft delete. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -429,8 +430,7 @@ def execute(self, context: Context) -> None: class CloudDataTransferServiceGetOperationOperator(GoogleCloudBaseOperator): """ - Gets the latest state of a long-running operation in Google Storage Transfer - Service. + Gets the latest state of a long-running operation in Google Storage Transfer Service. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -503,8 +503,7 @@ def execute(self, context: Context) -> dict: class CloudDataTransferServiceListOperationsOperator(GoogleCloudBaseOperator): """ - Lists long-running operations in Google Storage Transfer - Service that match the specified filter. + Lists long-running operations in Google Storage Transfer Service that match the specified filter. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -766,8 +765,7 @@ def execute(self, context: Context) -> None: class CloudDataTransferServiceS3ToGCSOperator(GoogleCloudBaseOperator): """ - Synchronizes an S3 bucket with a Google Cloud Storage bucket using the - Google Cloud Storage Transfer Service. + Sync an S3 bucket with a Google Cloud Storage bucket using the Google Cloud Storage Transfer Service. .. warning:: diff --git a/airflow/providers/google/cloud/operators/compute.py b/airflow/providers/google/cloud/operators/compute.py index db8a98e443484..2adb5b8439c4c 100644 --- a/airflow/providers/google/cloud/operators/compute.py +++ b/airflow/providers/google/cloud/operators/compute.py @@ -667,8 +667,7 @@ def execute(self, context: Context) -> None: class ComputeEngineSetMachineTypeOperator(ComputeEngineBaseOperator): """ - Changes the machine type for a stopped instance to the machine type specified in - the request. + Changes the machine type for a stopped instance to the machine type specified in the request. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -1261,9 +1260,10 @@ def execute(self, context: Context) -> dict: class ComputeEngineInstanceGroupUpdateManagerTemplateOperator(ComputeEngineBaseOperator): """ - Patches the Instance Group Manager, replacing source template URL with the - destination one. API V1 does not have update/patch operations for Instance - Group Manager, so you must use beta or newer API version. Beta is the default. + Patches the Instance Group Manager, replacing source template URL with the destination one. + + API V1 does not have update/patch operations for Instance Group Manager, + so you must use beta or newer API version. Beta is the default. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -1409,6 +1409,7 @@ def execute(self, context: Context) -> bool | None: class ComputeEngineInsertInstanceGroupManagerOperator(ComputeEngineBaseOperator): """ Creates an Instance Group Managers using the body specified. + After the group is created, instances in the group are created using the specified Instance Template. .. seealso:: @@ -1576,8 +1577,7 @@ def execute(self, context: Context) -> dict: class ComputeEngineDeleteInstanceGroupManagerOperator(ComputeEngineBaseOperator): """ - Deletes an Instance Group Managers. - Deleting an Instance Group Manager is permanent and cannot be undone. + Permanently and irrevocably deletes an Instance Group Managers. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/operators/dataflow.py b/airflow/providers/google/cloud/operators/dataflow.py index 5ae1115a34ae6..fa0e09a544896 100644 --- a/airflow/providers/google/cloud/operators/dataflow.py +++ b/airflow/providers/google/cloud/operators/dataflow.py @@ -60,9 +60,12 @@ class CheckJobRunning(Enum): class DataflowConfiguration: - """Dataflow configuration that can be passed to - :class:`~airflow.providers.apache.beam.operators.beam.BeamRunJavaPipelineOperator` and - :class:`~airflow.providers.apache.beam.operators.beam.BeamRunPythonPipelineOperator`. + """ + Dataflow configuration for BeamRunJavaPipelineOperator and BeamRunPythonPipelineOperator. + + .. seealso:: + :class:`~airflow.providers.apache.beam.operators.beam.BeamRunJavaPipelineOperator` + and :class:`~airflow.providers.apache.beam.operators.beam.BeamRunPythonPipelineOperator`. :param job_name: The 'jobName' to use when executing the Dataflow job (templated). This ends up being set in the pipeline options, so any entry @@ -166,11 +169,10 @@ def __init__( class DataflowCreateJavaJobOperator(GoogleCloudBaseOperator): """ - Start a Java Cloud Dataflow batch job. The parameters of the operation - will be passed to the job. + Start a Java Cloud Dataflow batch job; the parameters of the operation will be passed to the job. This class is deprecated. - Please use `providers.apache.beam.operators.beam.BeamRunJavaPipelineOperator`. + Please use :class:`providers.apache.beam.operators.beam.BeamRunJavaPipelineOperator`. Example usage: @@ -452,8 +454,7 @@ def on_kill(self) -> None: class DataflowTemplatedJobStartOperator(GoogleCloudBaseOperator): """ - Start a Templated Cloud Dataflow job. The parameters of the operation - will be passed to the job. + Start a Templated Cloud Dataflow job; the parameters of the operation will be passed to the job. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -1015,14 +1016,14 @@ def on_kill(self) -> None: class DataflowCreatePythonJobOperator(GoogleCloudBaseOperator): """ - Launching Cloud Dataflow jobs written in python. Note that both - dataflow_default_options and options will be merged to specify pipeline - execution parameter, and dataflow_default_options is expected to save - high-level options, for instances, project and zone information, which - apply to all dataflow operators in the DAG. + Launching Cloud Dataflow jobs written in python. + + Note that both dataflow_default_options and options will be merged to specify pipeline + execution parameter, and dataflow_default_options is expected to save high-level options, + for instances, project and zone information, which apply to all dataflow operators in the DAG. This class is deprecated. - Please use `providers.apache.beam.operators.beam.BeamRunPythonPipelineOperator`. + Please use :class:`providers.apache.beam.operators.beam.BeamRunPythonPipelineOperator`. .. seealso:: For more detail on job submission have a look at the reference: @@ -1232,6 +1233,7 @@ def on_kill(self) -> None: class DataflowStopJobOperator(GoogleCloudBaseOperator): """ Stops the job with the specified name prefix or Job ID. + All jobs with provided name prefix will be stopped. Streaming jobs are drained by default. diff --git a/airflow/providers/google/cloud/operators/datafusion.py b/airflow/providers/google/cloud/operators/datafusion.py index b214c36d7844d..ef32bfc821355 100644 --- a/airflow/providers/google/cloud/operators/datafusion.py +++ b/airflow/providers/google/cloud/operators/datafusion.py @@ -50,6 +50,7 @@ def get_project_id(instance): class CloudDataFusionRestartInstanceOperator(GoogleCloudBaseOperator): """ Restart a single Data Fusion instance. + At the end of an operation instance is fully restarted. .. seealso:: @@ -848,8 +849,8 @@ def execute(self, context: Context) -> str: def execute_complete(self, context: Context, event: dict[str, Any]): """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ if event["status"] == "error": raise AirflowException(event["message"]) diff --git a/airflow/providers/google/cloud/operators/dataproc.py b/airflow/providers/google/cloud/operators/dataproc.py index db7d785347a83..4e56aa2a25941 100644 --- a/airflow/providers/google/cloud/operators/dataproc.py +++ b/airflow/providers/google/cloud/operators/dataproc.py @@ -387,6 +387,7 @@ def _build_cluster_data(self): def make(self): """ Helper method for easier migration. + :return: Dict representing Dataproc cluster. """ return self._build_cluster_data() @@ -654,8 +655,8 @@ def execute(self, context: Context) -> dict: def execute_complete(self, context: Context, event: dict[str, Any]) -> Any: """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ cluster_state = event["cluster_state"] cluster_name = event["cluster_name"] @@ -894,8 +895,8 @@ def execute(self, context: Context) -> None: def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> Any: """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ if event and event["status"] == "error": raise AirflowException(event["message"]) @@ -1073,8 +1074,8 @@ def execute(self, context: Context): def execute_complete(self, context, event=None) -> None: """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ job_state = event["job_state"] job_id = event["job_id"] @@ -1086,10 +1087,7 @@ def execute_complete(self, context, event=None) -> None: return job_id def on_kill(self) -> None: - """ - Callback called when the operator is killed. - Cancel any running job. - """ + """Callback called when the operator is killed; cancel any running job.""" if self.dataproc_job_id: self.hook.cancel_job(project_id=self.project_id, job_id=self.dataproc_job_id, region=self.region) @@ -2200,8 +2198,8 @@ def execute(self, context: Context): def execute_complete(self, context: Context, event: dict[str, Any]) -> Any: """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ cluster_state = event["cluster_state"] cluster_name = event["cluster_name"] diff --git a/airflow/providers/google/cloud/operators/dlp.py b/airflow/providers/google/cloud/operators/dlp.py index dde0ce59438a1..60b52a4751c8f 100644 --- a/airflow/providers/google/cloud/operators/dlp.py +++ b/airflow/providers/google/cloud/operators/dlp.py @@ -15,11 +15,8 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -""" -This module contains various Google Cloud DLP operators -which allow you to perform basic operations using -Cloud DLP. -""" +"""Various Google Cloud DLP operators which allow you to perform basic operations using Cloud DLP.""" + from __future__ import annotations from typing import TYPE_CHECKING, Sequence @@ -152,8 +149,7 @@ def execute(self, context: Context) -> None: class CloudDLPCreateDeidentifyTemplateOperator(GoogleCloudBaseOperator): """ - Creates a DeidentifyTemplate for re-using frequently used configuration for - de-identifying content, images, and storage. + Create a deidentify template to reuse frequently-used configurations for content, images, and storage. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -374,8 +370,7 @@ def execute(self, context: Context): class CloudDLPCreateInspectTemplateOperator(GoogleCloudBaseOperator): """ - Creates an InspectTemplate for re-using frequently used configuration for - inspecting content, images, and storage. + Create an InspectTemplate to reuse frequently-used configurations for content, images, and storage. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -485,8 +480,7 @@ def execute(self, context: Context): class CloudDLPCreateJobTriggerOperator(GoogleCloudBaseOperator): """ - Creates a job trigger to run DLP actions such as scanning storage for sensitive - information on a set schedule. + Create a job trigger to run DLP actions such as scanning storage for sensitive info on a set schedule. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -705,8 +699,7 @@ def execute(self, context: Context): class CloudDLPDeidentifyContentOperator(GoogleCloudBaseOperator): """ - De-identifies potentially sensitive info from a ContentItem. This method has limits - on input size and output size. + De-identifies potentially sensitive info from a content item; limits input size and output size. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -893,8 +886,10 @@ def execute(self, context: Context) -> None: class CloudDLPDeleteDLPJobOperator(GoogleCloudBaseOperator): """ - Deletes a long-running DlpJob. This method indicates that the client is no longer - interested in the DlpJob result. The job will be cancelled if possible. + Deletes a long-running DlpJob. + + This method indicates that the client is no longer interested + in the DlpJob result. The job will be cancelled if possible. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -1676,8 +1671,7 @@ def execute(self, context: Context): class CloudDLPInspectContentOperator(GoogleCloudBaseOperator): """ - Finds potentially sensitive info in content. This method has limits on - input size, processing time, and output size. + Finds potentially sensitive info in content; limits input size, processing time, and output size. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -2317,8 +2311,7 @@ def execute(self, context: Context): class CloudDLPRedactImageOperator(GoogleCloudBaseOperator): """ - Redacts potentially sensitive info from an image. This method has limits on - input size, processing time, and output size. + Redacts potentially sensitive info from an image; limits input size, processing time, and output size. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/operators/functions.py b/airflow/providers/google/cloud/operators/functions.py index 1f01880de4b78..1131790cabdf5 100644 --- a/airflow/providers/google/cloud/operators/functions.py +++ b/airflow/providers/google/cloud/operators/functions.py @@ -104,8 +104,7 @@ def _validate_max_instances(value): class CloudFunctionDeployFunctionOperator(GoogleCloudBaseOperator): """ - Creates a function in Google Cloud Functions. - If a function with this name already exists, it will be updated. + Create or update a function in Google Cloud Functions. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -312,10 +311,7 @@ def should_upload_function(self) -> bool: return self.upload_function def preprocess_body(self) -> None: - """ - Modifies sourceUploadUrl body field in special way when zip_path - is not empty. - """ + """Modifies sourceUploadUrl body field in special way when zip_path is not empty.""" self._verify_archive_url_and_zip_path() self._verify_upload_url_and_zip_path() self._verify_upload_url_and_no_zip_path() @@ -412,8 +408,7 @@ def execute(self, context: Context): class CloudFunctionInvokeFunctionOperator(GoogleCloudBaseOperator): """ - Invokes a deployed Cloud Function. To be used for testing - purposes as very limited traffic is allowed. + Invokes a deployed Cloud Function. To be used for testing purposes as very limited traffic is allowed. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/operators/gcs.py b/airflow/providers/google/cloud/operators/gcs.py index 6e50cbad2d16e..47ece7e412a43 100644 --- a/airflow/providers/google/cloud/operators/gcs.py +++ b/airflow/providers/google/cloud/operators/gcs.py @@ -43,8 +43,10 @@ class GCSCreateBucketOperator(GoogleCloudBaseOperator): """ - Creates a new bucket. Google Cloud Storage uses a flat namespace, - so you can't create a bucket with a name that is already in use. + Creates a new bucket. + + Google Cloud Storage uses a flat namespace, so you + can't create a bucket with a name that is already in use. .. seealso:: For more information, see Bucket Naming Guidelines: @@ -158,8 +160,7 @@ def execute(self, context: Context) -> None: class GCSListObjectsOperator(GoogleCloudBaseOperator): """ - List all objects from the bucket filtered by given string prefix and delimiter in name, - or match_glob. + List all objects from the bucket filtered by given string prefix and delimiter in name or match_glob. This operator returns a python list with the name of objects which can be used by XCom in the downstream task. @@ -265,9 +266,7 @@ def execute(self, context: Context) -> list: class GCSDeleteObjectsOperator(GoogleCloudBaseOperator): """ - Deletes objects from a Google Cloud Storage bucket, either - from an explicit list of object names or all objects - matching a prefix. + Deletes objects from a list or all objects matching a prefix from a Google Cloud Storage bucket. :param bucket_name: The GCS bucket to delete from :param objects: List of objects to delete. These should be the names @@ -495,11 +494,11 @@ def execute(self, context: Context) -> None: class GCSFileTransformOperator(GoogleCloudBaseOperator): """ - Copies data from a source GCS location to a temporary location on the - local filesystem. Runs a transformation on this file as specified by - the transformation script and uploads the output to a destination bucket. - If the output bucket is not specified the original file will be - overwritten. + Copies data from a source GCS location to a temporary location on the local filesystem. + + Runs a transformation on this file as specified by the transformation script + and uploads the output to a destination bucket. If the output bucket is not + specified the original file will be overwritten. The locations of the source and the destination files in the local filesystem is provided as an first and second arguments to the @@ -601,6 +600,8 @@ def execute(self, context: Context) -> None: class GCSTimeSpanFileTransformOperator(GoogleCloudBaseOperator): """ + Copy objects that were modified during a time span, run a transform, and upload results to a bucket. + Determines a list of objects that were added or modified at a GCS source location during a specific time-span, copies them to a temporary location on the local file system, runs a transform on this file as specified by diff --git a/airflow/providers/google/cloud/operators/kubernetes_engine.py b/airflow/providers/google/cloud/operators/kubernetes_engine.py index bf14828d87027..f0d991f10dfac 100644 --- a/airflow/providers/google/cloud/operators/kubernetes_engine.py +++ b/airflow/providers/google/cloud/operators/kubernetes_engine.py @@ -184,8 +184,7 @@ def _get_hook(self) -> GKEHook: class GKECreateClusterOperator(GoogleCloudBaseOperator): """ - Create a Google Kubernetes Engine Cluster of specified dimensions - The operator will wait until the cluster is created. + Create a Google Kubernetes Engine Cluster of specified dimensions and wait until the cluster is created. The **minimum** required to define a cluster to create is: @@ -392,8 +391,7 @@ def _get_hook(self) -> GKEHook: class GKEStartPodOperator(KubernetesPodOperator): """ - Executes a task in a Kubernetes pod in the specified Google Kubernetes - Engine cluster. + Executes a task in a Kubernetes pod in the specified Google Kubernetes Engine cluster. This Operator assumes that the system has gcloud installed and has configured a connection id with a service account. diff --git a/airflow/providers/google/cloud/operators/mlengine.py b/airflow/providers/google/cloud/operators/mlengine.py index b776d20dff42d..7c375be519f00 100644 --- a/airflow/providers/google/cloud/operators/mlengine.py +++ b/airflow/providers/google/cloud/operators/mlengine.py @@ -1299,8 +1299,8 @@ def _wait_for_job_done(self, project_id: str, job_id: str, interval: int = 30): def execute_complete(self, context: Context, event: dict[str, Any]): """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ if event["status"] == "error": raise AirflowException(event["message"]) diff --git a/airflow/providers/google/cloud/operators/natural_language.py b/airflow/providers/google/cloud/operators/natural_language.py index 2665f9ced60ab..21ddb01f8a894 100644 --- a/airflow/providers/google/cloud/operators/natural_language.py +++ b/airflow/providers/google/cloud/operators/natural_language.py @@ -37,8 +37,9 @@ class CloudNaturalLanguageAnalyzeEntitiesOperator(GoogleCloudBaseOperator): """ - Finds named entities in the text along with entity types, - salience, mentions for each entity, and other properties. + Finds named entities in the text along with various properties. + + Examples properties: entity types, salience, mentions for each entity, and others. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -109,8 +110,7 @@ def execute(self, context: Context): class CloudNaturalLanguageAnalyzeEntitySentimentOperator(GoogleCloudBaseOperator): """ - Finds entities, similar to AnalyzeEntities in the text and analyzes sentiment associated with each - entity and its mentions. + Similar to AnalyzeEntities, also analyzes sentiment associated with each entity and its mentions. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/operators/pubsub.py b/airflow/providers/google/cloud/operators/pubsub.py index 2eb6882ceb992..8e0f7a12d99a3 100644 --- a/airflow/providers/google/cloud/operators/pubsub.py +++ b/airflow/providers/google/cloud/operators/pubsub.py @@ -678,7 +678,9 @@ def execute(self, context: Context) -> None: class PubSubPullOperator(GoogleCloudBaseOperator): - """Pulls messages from a PubSub subscription and passes them through XCom. + """ + Pulls messages from a PubSub subscription and passes them through XCom. + If the queue is empty, returns empty list - never waits for messages. If you do need to wait, please use :class:`airflow.providers.google.cloud.sensors.PubSubPullSensor` instead. @@ -781,6 +783,7 @@ def _default_message_callback( ) -> list: """ This method can be overridden by subclasses or by `messages_callback` constructor argument. + This default implementation converts `ReceivedMessage` objects into JSON-serializable dicts. :param pulled_messages: messages received from the topic. diff --git a/airflow/providers/google/cloud/operators/spanner.py b/airflow/providers/google/cloud/operators/spanner.py index 72a2e7000d8e0..d235c146b5b48 100644 --- a/airflow/providers/google/cloud/operators/spanner.py +++ b/airflow/providers/google/cloud/operators/spanner.py @@ -31,8 +31,7 @@ class SpannerDeployInstanceOperator(GoogleCloudBaseOperator): """ - Creates a new Cloud Spanner instance, or if an instance with the same instance_id - exists in the specified project, updates the Cloud Spanner instance. + Create or update a Cloud Spanner instance. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -128,8 +127,7 @@ def execute(self, context: Context) -> None: class SpannerDeleteInstanceOperator(GoogleCloudBaseOperator): """ - Deletes a Cloud Spanner instance. If an instance does not exist, - no action is taken and the operator succeeds. + Delete a Cloud Spanner instance; if an instance does not exist, no action is taken and the task succeeds. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -309,8 +307,7 @@ def sanitize_queries(queries: list[str]) -> None: class SpannerDeployDatabaseInstanceOperator(GoogleCloudBaseOperator): """ - Creates a new Cloud Spanner database, or if database exists, - the operator does nothing. + Creates a new Cloud Spanner database; if database exists, the operator does nothing. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/operators/stackdriver.py b/airflow/providers/google/cloud/operators/stackdriver.py index 33a082355e0db..aa2b3d368c678 100644 --- a/airflow/providers/google/cloud/operators/stackdriver.py +++ b/airflow/providers/google/cloud/operators/stackdriver.py @@ -36,11 +36,11 @@ class StackdriverListAlertPoliciesOperator(GoogleCloudBaseOperator): """ - Fetches all the Alert Policies identified by the filter passed as - filter parameter. The desired return type can be specified by the - format parameter, the supported formats are "dict", "json" and None - which returns python dictionary, stringified JSON and protobuf - respectively. + Fetches all the Alert Policies identified by the filter passed as filter parameter. + + The desired return type can be specified by the format parameter, the supported + formats are "dict", "json" and None which returns python dictionary, stringified + JSON and protobuf respectively. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -150,8 +150,9 @@ def execute(self, context: Context): class StackdriverEnableAlertPoliciesOperator(GoogleCloudBaseOperator): """ - Enables one or more disabled alerting policies identified by filter - parameter. Inoperative in case the policy is already enabled. + Enables one or more disabled alerting policies identified by filter parameter. + + Inoperative in case the policy is already enabled. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -232,8 +233,9 @@ def execute(self, context: Context): # Disable Alert Operator class StackdriverDisableAlertPoliciesOperator(GoogleCloudBaseOperator): """ - Disables one or more enabled alerting policies identified by filter - parameter. Inoperative in case the policy is already disabled. + Disables one or more enabled alerting policies identified by filter parameter. + + Inoperative in case the policy is already disabled. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -313,8 +315,7 @@ def execute(self, context: Context): class StackdriverUpsertAlertOperator(GoogleCloudBaseOperator): """ - Creates a new alert or updates an existing policy identified - the name field in the alerts parameter. + Creates a new alert or updates an existing policy identified the name field in the alerts parameter. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -470,11 +471,11 @@ def execute(self, context: Context): class StackdriverListNotificationChannelsOperator(GoogleCloudBaseOperator): """ - Fetches all the Notification Channels identified by the filter passed as - filter parameter. The desired return type can be specified by the - format parameter, the supported formats are "dict", "json" and None - which returns python dictionary, stringified JSON and protobuf - respectively. + Fetches all the Notification Channels identified by the filter passed as filter parameter. + + The desired return type can be specified by the format parameter, the + supported formats are "dict", "json" and None which returns python + dictionary, stringified JSON and protobuf respectively. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -584,8 +585,9 @@ def execute(self, context: Context): class StackdriverEnableNotificationChannelsOperator(GoogleCloudBaseOperator): """ - Enables one or more disabled alerting policies identified by filter - parameter. Inoperative in case the policy is already enabled. + Enables one or more disabled alerting policies identified by filter parameter. + + Inoperative in case the policy is already enabled. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -668,8 +670,9 @@ def execute(self, context: Context): class StackdriverDisableNotificationChannelsOperator(GoogleCloudBaseOperator): """ - Disables one or more enabled notification channels identified by filter - parameter. Inoperative in case the policy is already disabled. + Disables one or more enabled notification channels identified by filter parameter. + + Inoperative in case the policy is already disabled. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -752,8 +755,9 @@ def execute(self, context: Context): class StackdriverUpsertNotificationChannelOperator(GoogleCloudBaseOperator): """ - Creates a new notification or updates an existing notification channel - identified the name field in the alerts parameter. + Create a new notification or updates an existing notification channel. + + Channel is identified by the name field in the alerts parameter. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/operators/tasks.py b/airflow/providers/google/cloud/operators/tasks.py index f0964214e2c84..f78399d766734 100644 --- a/airflow/providers/google/cloud/operators/tasks.py +++ b/airflow/providers/google/cloud/operators/tasks.py @@ -15,11 +15,8 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -""" -This module contains various Google Cloud Tasks operators -which allow you to perform basic operations using -Cloud Tasks queues/tasks. -""" +"""Google Cloud Tasks operators which allow you to perform basic operations using Cloud Tasks queues/tasks.""" + from __future__ import annotations from typing import TYPE_CHECKING, Sequence, Tuple diff --git a/airflow/providers/google/cloud/operators/vertex_ai/auto_ml.py b/airflow/providers/google/cloud/operators/vertex_ai/auto_ml.py index a3bcb2158d340..88a8560e85dc2 100644 --- a/airflow/providers/google/cloud/operators/vertex_ai/auto_ml.py +++ b/airflow/providers/google/cloud/operators/vertex_ai/auto_ml.py @@ -80,10 +80,7 @@ def __init__( self.hook: AutoMLHook | None = None def on_kill(self) -> None: - """ - Callback called when the operator is killed. - Cancel any running job. - """ + """Callback called when the operator is killed; cancel any running job.""" if self.hook: self.hook.cancel_auto_ml_job() @@ -537,8 +534,11 @@ def execute(self, context: Context): class DeleteAutoMLTrainingJobOperator(GoogleCloudBaseOperator): - """Deletes an AutoMLForecastingTrainingJob, AutoMLImageTrainingJob, AutoMLTabularTrainingJob, - AutoMLTextTrainingJob, or AutoMLVideoTrainingJob. + """ + Delete an AutoML training job. + + Can be used with AutoMLForecastingTrainingJob, AutoMLImageTrainingJob, + AutoMLTabularTrainingJob, AutoMLTextTrainingJob, or AutoMLVideoTrainingJob. """ template_fields = ("training_pipeline", "region", "project_id", "impersonation_chain") @@ -588,7 +588,10 @@ def execute(self, context: Context): class ListAutoMLTrainingJobOperator(GoogleCloudBaseOperator): - """Lists AutoMLForecastingTrainingJob, AutoMLImageTrainingJob, AutoMLTabularTrainingJob, + """ + List an AutoML training job. + + Can be used with AutoMLForecastingTrainingJob, AutoMLImageTrainingJob, AutoMLTabularTrainingJob, AutoMLTextTrainingJob, or AutoMLVideoTrainingJob in a Location. """ diff --git a/airflow/providers/google/cloud/operators/vertex_ai/batch_prediction_job.py b/airflow/providers/google/cloud/operators/vertex_ai/batch_prediction_job.py index 0247d33f84404..684cfdd8f2e8d 100644 --- a/airflow/providers/google/cloud/operators/vertex_ai/batch_prediction_job.py +++ b/airflow/providers/google/cloud/operators/vertex_ai/batch_prediction_job.py @@ -268,10 +268,7 @@ def execute(self, context: Context): return batch_prediction_job def on_kill(self) -> None: - """ - Callback called when the operator is killed. - Cancel any running job. - """ + """Callback called when the operator is killed; cancel any running job.""" if self.hook: self.hook.cancel_batch_prediction_job() diff --git a/airflow/providers/google/cloud/operators/vertex_ai/custom_job.py b/airflow/providers/google/cloud/operators/vertex_ai/custom_job.py index 7eb7866ff3671..9896cdcccd693 100644 --- a/airflow/providers/google/cloud/operators/vertex_ai/custom_job.py +++ b/airflow/providers/google/cloud/operators/vertex_ai/custom_job.py @@ -489,10 +489,7 @@ def execute(self, context: Context): return result def on_kill(self) -> None: - """ - Callback called when the operator is killed. - Cancel any running job. - """ + """Callback called when the operator is killed; cancel any running job.""" if self.hook: self.hook.cancel_job() @@ -839,10 +836,7 @@ def execute(self, context: Context): return result def on_kill(self) -> None: - """ - Callback called when the operator is killed. - Cancel any running job. - """ + """Callback called when the operator is killed; cancel any running job.""" if self.hook: self.hook.cancel_job() @@ -1191,16 +1185,14 @@ def execute(self, context: Context): return result def on_kill(self) -> None: - """ - Callback called when the operator is killed. - Cancel any running job. - """ + """Callback called when the operator is killed; cancel any running job.""" if self.hook: self.hook.cancel_job() class DeleteCustomTrainingJobOperator(GoogleCloudBaseOperator): - """Deletes a CustomTrainingJob, CustomPythonTrainingJob, or CustomContainerTrainingJob. + """ + Deletes a CustomTrainingJob, CustomPythonTrainingJob, or CustomContainerTrainingJob. :param training_pipeline_id: Required. The name of the TrainingPipeline resource to be deleted. :param custom_job_id: Required. The name of the CustomJob to delete. diff --git a/airflow/providers/google/cloud/operators/vertex_ai/endpoint_service.py b/airflow/providers/google/cloud/operators/vertex_ai/endpoint_service.py index 10c722b2a0e23..b2bf7c4213391 100644 --- a/airflow/providers/google/cloud/operators/vertex_ai/endpoint_service.py +++ b/airflow/providers/google/cloud/operators/vertex_ai/endpoint_service.py @@ -458,8 +458,7 @@ def execute(self, context: Context): class UndeployModelOperator(GoogleCloudBaseOperator): """ - Undeploys a Model from an Endpoint, removing a DeployedModel from it, and freeing all resources it's - using. + Undeploys a Model from an Endpoint, removing a DeployedModel from it, and freeing all used resources. :param project_id: Required. The ID of the Google Cloud project that the service belongs to. :param region: Required. The ID of the Google Cloud region that the service belongs to. diff --git a/airflow/providers/google/cloud/operators/vertex_ai/hyperparameter_tuning_job.py b/airflow/providers/google/cloud/operators/vertex_ai/hyperparameter_tuning_job.py index c3d6c82b86d3f..69c7e76016d31 100644 --- a/airflow/providers/google/cloud/operators/vertex_ai/hyperparameter_tuning_job.py +++ b/airflow/providers/google/cloud/operators/vertex_ai/hyperparameter_tuning_job.py @@ -264,10 +264,7 @@ def execute(self, context: Context): return hyperparameter_tuning_job def on_kill(self) -> None: - """ - Callback called when the operator is killed. - Cancel any running job. - """ + """Callback called when the operator is killed; cancel any running job.""" if self.hook: self.hook.cancel_hyperparameter_tuning_job() diff --git a/airflow/providers/google/cloud/operators/workflows.py b/airflow/providers/google/cloud/operators/workflows.py index cff6e15ef0b27..4f2517f5da017 100644 --- a/airflow/providers/google/cloud/operators/workflows.py +++ b/airflow/providers/google/cloud/operators/workflows.py @@ -49,9 +49,10 @@ class WorkflowsCreateWorkflowOperator(GoogleCloudBaseOperator): """ - Creates a new workflow. If a workflow with the specified name - already exists in the specified project and location, the long - running operation will return + Creates a new workflow. + + If a workflow with the specified name already exists in the specified + project and location, the long running operation will return [ALREADY_EXISTS][google.rpc.Code.ALREADY_EXISTS] error. .. seealso:: @@ -159,6 +160,7 @@ def execute(self, context: Context): class WorkflowsUpdateWorkflowOperator(GoogleCloudBaseOperator): """ Updates an existing workflow. + Running this method has no impact on already running executions of the workflow. A new revision of the workflow may be created as a result of a successful @@ -245,9 +247,7 @@ def execute(self, context: Context): class WorkflowsDeleteWorkflowOperator(GoogleCloudBaseOperator): """ - Deletes a workflow with the specified name. - This method also cancels and deletes all running - executions of the workflow. + Delete a workflow with the specified name and all running executions of the workflow. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -305,8 +305,7 @@ def execute(self, context: Context): class WorkflowsListWorkflowsOperator(GoogleCloudBaseOperator): """ - Lists Workflows in a given project and location. - The default order is not specified. + Lists Workflows in a given project and location; the default order is not specified. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -447,8 +446,7 @@ def execute(self, context: Context): class WorkflowsCreateExecutionOperator(GoogleCloudBaseOperator): """ - Creates a new execution using the latest revision of - the given workflow. + Creates a new execution using the latest revision of the given workflow. .. seealso:: For more information on how to use this operator, take a look at the guide: @@ -597,11 +595,10 @@ def execute(self, context: Context): class WorkflowsListExecutionsOperator(GoogleCloudBaseOperator): """ - Returns a list of executions which belong to the - workflow with the given name. The method returns - executions of all workflow revisions. Returned - executions are ordered by their start time (newest - first). + Returns a list of executions which belong to the workflow with the given name. + + The method returns executions of all workflow revisions. Returned + executions are ordered by their start time (newest first). .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/sensors/bigquery.py b/airflow/providers/google/cloud/sensors/bigquery.py index db109bf2c1e89..0df6ee06a0bc7 100644 --- a/airflow/providers/google/cloud/sensors/bigquery.py +++ b/airflow/providers/google/cloud/sensors/bigquery.py @@ -132,8 +132,8 @@ def execute(self, context: Context) -> None: def execute_complete(self, context: dict[str, Any], event: dict[str, str] | None = None) -> str: """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ table_uri = f"{self.project_id}:{self.dataset_id}.{self.table_id}" self.log.info("Sensor checks existence of table: %s", table_uri) @@ -215,10 +215,7 @@ def poke(self, context: Context) -> bool: ) def execute(self, context: Context) -> None: - """ - Airflow runs this method on the worker and defers using the triggers - if deferrable is set to True. - """ + """Airflow runs this method on the worker and defers using the triggers if deferrable is True.""" if not self.deferrable: super().execute(context) else: @@ -242,8 +239,8 @@ def execute(self, context: Context) -> None: def execute_complete(self, context: dict[str, Any], event: dict[str, str] | None = None) -> str: """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ table_uri = f"{self.project_id}:{self.dataset_id}.{self.table_id}" self.log.info('Sensor checks existence of partition: "%s" in table: %s', self.partition_id, table_uri) diff --git a/airflow/providers/google/cloud/sensors/bigtable.py b/airflow/providers/google/cloud/sensors/bigtable.py index c69d4099cad1c..0b823f0a1a07d 100644 --- a/airflow/providers/google/cloud/sensors/bigtable.py +++ b/airflow/providers/google/cloud/sensors/bigtable.py @@ -36,6 +36,7 @@ class BigtableTableReplicationCompletedSensor(BaseSensorOperator, BigtableValidationMixin): """ Sensor that waits for Cloud Bigtable table to be fully replicated to its clusters. + No exception will be raised if the instance or the table does not exist. For more details about cluster states for a table, have a look at the reference: diff --git a/airflow/providers/google/cloud/sensors/cloud_composer.py b/airflow/providers/google/cloud/sensors/cloud_composer.py index 238c9533051c3..ecd717aa5a152 100644 --- a/airflow/providers/google/cloud/sensors/cloud_composer.py +++ b/airflow/providers/google/cloud/sensors/cloud_composer.py @@ -84,8 +84,8 @@ def execute(self, context: Context) -> None: def execute_complete(self, context: dict[str, Any], event: dict[str, str] | None = None) -> str: """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ if event: if event.get("operation_done"): diff --git a/airflow/providers/google/cloud/sensors/cloud_storage_transfer_service.py b/airflow/providers/google/cloud/sensors/cloud_storage_transfer_service.py index cccc628d3c60e..60edca6337d1f 100644 --- a/airflow/providers/google/cloud/sensors/cloud_storage_transfer_service.py +++ b/airflow/providers/google/cloud/sensors/cloud_storage_transfer_service.py @@ -35,8 +35,7 @@ class CloudDataTransferServiceJobStatusSensor(BaseSensorOperator): """ - Waits for at least one operation belonging to the job to have the - expected status. + Waits for at least one operation belonging to the job to have the expected status. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/sensors/dataproc_metastore.py b/airflow/providers/google/cloud/sensors/dataproc_metastore.py index 50ba4ddf55eee..fb895abba33dc 100644 --- a/airflow/providers/google/cloud/sensors/dataproc_metastore.py +++ b/airflow/providers/google/cloud/sensors/dataproc_metastore.py @@ -33,7 +33,8 @@ class MetastoreHivePartitionSensor(BaseSensorOperator): """ Waits for partitions to show up in Hive. - This sensor uses Google Cloud SDK and passes requests via gRPC. + + This sensor uses Google Cloud SDK and passes requests via gRPC. :param service_id: Required. Dataproc Metastore service id. :param region: Required. The ID of the Google Cloud region that the service belongs to. diff --git a/airflow/providers/google/cloud/sensors/gcs.py b/airflow/providers/google/cloud/sensors/gcs.py index 08fd37022dfe7..5542dc88af46c 100644 --- a/airflow/providers/google/cloud/sensors/gcs.py +++ b/airflow/providers/google/cloud/sensors/gcs.py @@ -120,8 +120,8 @@ def execute(self, context: Context) -> None: def execute_complete(self, context: Context, event: dict[str, str]) -> str: """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ if event["status"] == "error": raise AirflowException(event["message"]) @@ -160,10 +160,10 @@ def __init__(self, **kwargs: Any) -> None: def ts_function(context): """ - Default callback for the GoogleCloudStorageObjectUpdatedSensor. The default - behaviour is check for the object being updated after the data interval's - end, or execution_date + interval on Airflow versions prior to 2.2 (before - AIP-39 implementation). + Default callback for the GoogleCloudStorageObjectUpdatedSensor. + + The default behaviour is check for the object being updated after the data interval's end, + or execution_date + interval on Airflow versions prior to 2.2 (before AIP-39 implementation). """ try: return context["data_interval_end"] @@ -341,10 +341,7 @@ def execute(self, context: Context): ) def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[str]]) -> str | list[str]: - """ - Callback for when the trigger fires; returns immediately. - Relies on trigger to throw a success event. - """ + """Callback for the trigger; returns immediately and relies on trigger to throw a success event.""" self.log.info("Resuming from trigger and checking status") if event["status"] == "success": return event["matches"] @@ -352,16 +349,15 @@ def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[ def get_time(): - """ - This is just a wrapper of datetime.datetime.now to simplify mocking in the - unittests. - """ + """This is just a wrapper of datetime.datetime.now to simplify mocking in the unittests.""" return datetime.now() @poke_mode_only class GCSUploadSessionCompleteSensor(BaseSensorOperator): """ + Return True if the inactivity period has passed with no increase in the number of objects in the bucket. + Checks for changes in the number of objects at prefix in Google Cloud Storage bucket and returns True if the inactivity period has passed with no increase in the number of objects. Note, this sensor will not behave correctly @@ -443,8 +439,7 @@ def _get_gcs_hook(self) -> GCSHook | None: def is_bucket_updated(self, current_objects: set[str]) -> bool: """ - Checks whether new objects have been uploaded and the inactivity_period - has passed and updates the state of the sensor accordingly. + Check whether new objects have been added and the inactivity_period has passed, and update the state. :param current_objects: set of object ids in bucket during last poke. """ @@ -547,8 +542,8 @@ def execute(self, context: Context) -> None: def execute_complete(self, context: dict[str, Any], event: dict[str, str] | None = None) -> str: """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ if event: if event["status"] == "success": diff --git a/airflow/providers/google/cloud/sensors/pubsub.py b/airflow/providers/google/cloud/sensors/pubsub.py index 2e03b3669d3dc..e82b20952d909 100644 --- a/airflow/providers/google/cloud/sensors/pubsub.py +++ b/airflow/providers/google/cloud/sensors/pubsub.py @@ -33,7 +33,9 @@ class PubSubPullSensor(BaseSensorOperator): - """Pulls messages from a PubSub subscription and passes them through XCom. + """ + Pulls messages from a PubSub subscription and passes them through XCom. + Always waits for at least one message to be returned from the subscription. .. seealso:: @@ -146,10 +148,7 @@ def poke(self, context: Context) -> bool: return bool(pulled_messages) def execute(self, context: Context) -> None: - """ - Airflow runs this method on the worker and defers using the triggers - if deferrable is set to True. - """ + """Airflow runs this method on the worker and defers using the triggers if deferrable is True.""" if not self.deferrable: super().execute(context) return self._return_value @@ -170,10 +169,7 @@ def execute(self, context: Context) -> None: ) def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[str]]) -> str | list[str]: - """ - Callback for when the trigger fires; returns immediately. - Relies on trigger to throw a success event. - """ + """Callback for the trigger; returns immediately and relies on trigger to throw a success event.""" if event["status"] == "success": self.log.info("Sensor pulls messages: %s", event["message"]) return event["message"] @@ -187,6 +183,7 @@ def _default_message_callback( ): """ This method can be overridden by subclasses or by `messages_callback` constructor argument. + This default implementation converts `ReceivedMessage` objects into JSON-serializable dicts. :param pulled_messages: messages received from the topic. diff --git a/airflow/providers/google/cloud/sensors/tasks.py b/airflow/providers/google/cloud/sensors/tasks.py index 714fcd18d630c..f90154e7feace 100644 --- a/airflow/providers/google/cloud/sensors/tasks.py +++ b/airflow/providers/google/cloud/sensors/tasks.py @@ -29,8 +29,7 @@ class TaskQueueEmptySensor(BaseSensorOperator): """ - Pulls tasks count from a cloud task queue. - Always waits for queue returning tasks count as 0. + Pulls tasks count from a cloud task queue; waits for queue to return task count as 0. :param project_id: the Google Cloud project ID for the subscription (templated) :param gcp_conn_id: The connection ID to use connecting to Google Cloud. diff --git a/airflow/providers/google/cloud/transfers/adls_to_gcs.py b/airflow/providers/google/cloud/transfers/adls_to_gcs.py index 4810ab41af20d..75b691ba41d2d 100644 --- a/airflow/providers/google/cloud/transfers/adls_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/adls_to_gcs.py @@ -15,10 +15,8 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -""" -This module contains Azure Data Lake Storage to -Google Cloud Storage operator. -""" +"""This module contains Azure Data Lake Storage to Google Cloud Storage operator.""" + from __future__ import annotations import os diff --git a/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py b/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py index 39d14bd048140..e8410cb5daa46 100644 --- a/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py @@ -31,8 +31,9 @@ class AzureFileShareToGCSOperator(BaseOperator): """ - Synchronizes a Azure FileShare directory content (excluding subdirectories), - possibly filtered by a prefix, with a Google Cloud Storage destination path. + Sync an Azure FileShare directory with a Google Cloud Storage destination path. + + Does not include subdirectories. May be filtered by prefix. :param share_name: The Azure FileShare share where to find the objects. (templated) :param directory_name: (Optional) Path to Azure FileShare directory which content is to be transferred. diff --git a/airflow/providers/google/cloud/transfers/bigquery_to_gcs.py b/airflow/providers/google/cloud/transfers/bigquery_to_gcs.py index 7ec62db9bfbd0..51ddfdd9d139b 100644 --- a/airflow/providers/google/cloud/transfers/bigquery_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/bigquery_to_gcs.py @@ -264,8 +264,8 @@ def execute(self, context: Context): def execute_complete(self, context: Context, event: dict[str, Any]): """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ if event["status"] == "error": raise AirflowException(event["message"]) diff --git a/airflow/providers/google/cloud/transfers/bigquery_to_mssql.py b/airflow/providers/google/cloud/transfers/bigquery_to_mssql.py index 0c7e58a458c03..d46a001c28a23 100644 --- a/airflow/providers/google/cloud/transfers/bigquery_to_mssql.py +++ b/airflow/providers/google/cloud/transfers/bigquery_to_mssql.py @@ -32,8 +32,7 @@ class BigQueryToMsSqlOperator(BigQueryToSqlBaseOperator): """ - Fetches the data from a BigQuery table (alternatively fetch data for selected columns) - and insert that data into a MSSQL table. + Fetch data from a BigQuery table (alternatively fetch selected columns) and insert it into a MSSQL table. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/transfers/bigquery_to_mysql.py b/airflow/providers/google/cloud/transfers/bigquery_to_mysql.py index 99b01fe9e86be..c8aec07b0e40b 100644 --- a/airflow/providers/google/cloud/transfers/bigquery_to_mysql.py +++ b/airflow/providers/google/cloud/transfers/bigquery_to_mysql.py @@ -28,8 +28,7 @@ class BigQueryToMySqlOperator(BigQueryToSqlBaseOperator): """ - Fetches the data from a BigQuery table (alternatively fetch data for selected columns) - and insert that data into a MySQL table. + Fetch data from a BigQuery table (alternatively fetch selected columns) and insert it into a MySQL table. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/transfers/bigquery_to_postgres.py b/airflow/providers/google/cloud/transfers/bigquery_to_postgres.py index 34d6399620476..3d1415687c68e 100644 --- a/airflow/providers/google/cloud/transfers/bigquery_to_postgres.py +++ b/airflow/providers/google/cloud/transfers/bigquery_to_postgres.py @@ -26,8 +26,7 @@ class BigQueryToPostgresOperator(BigQueryToSqlBaseOperator): """ - Fetches the data from a BigQuery table (alternatively fetch data for selected columns) - and insert that data into a PostgreSQL table. + Fetch data from a BigQuery table (alternatively fetch selected columns) and insert into PostgreSQL table. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/transfers/bigquery_to_sql.py b/airflow/providers/google/cloud/transfers/bigquery_to_sql.py index 0e3d7a2a69975..36acb9363811e 100644 --- a/airflow/providers/google/cloud/transfers/bigquery_to_sql.py +++ b/airflow/providers/google/cloud/transfers/bigquery_to_sql.py @@ -32,9 +32,10 @@ class BigQueryToSqlBaseOperator(BaseOperator): """ - Fetches the data from a BigQuery table (alternatively fetch data for selected columns) - and insert that data into a SQL table. This is a BaseOperator; an abstract class. Refer - to children classes which are related to specific SQL databases (MySQL, MsSQL, Postgres...). + Fetch data from a BigQuery table (alternatively fetch selected columns) and insert it into an SQL table. + + This is a BaseOperator; an abstract class. Refer to children classes + which are related to specific SQL databases (MySQL, MsSQL, Postgres...). .. note:: If you pass fields to ``selected_fields`` which are in different order than the diff --git a/airflow/providers/google/cloud/transfers/cassandra_to_gcs.py b/airflow/providers/google/cloud/transfers/cassandra_to_gcs.py index ffd3e3aa78373..408977d7ccfa4 100644 --- a/airflow/providers/google/cloud/transfers/cassandra_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/cassandra_to_gcs.py @@ -15,10 +15,8 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -""" -This module contains operator for copying -data from Cassandra to Google Cloud Storage in JSON format. -""" +"""This module contains operator for copying data from Cassandra to Google Cloud Storage in JSON format.""" + from __future__ import annotations import json @@ -220,8 +218,7 @@ def _write_local_data_files(self, cursor): def _write_local_schema_file(self, cursor): """ - Takes a cursor, and writes the BigQuery schema for the results to a - local file system. + Takes a cursor, and writes the BigQuery schema for the results to a local file system. :return: A dictionary where key is a filename to be used as an object name in GCS, and values are file handles to local files that @@ -295,9 +292,9 @@ def convert_array_types(self, value: list[Any] | SortedSet) -> list[Any]: def convert_user_type(self, value: Any) -> dict[str, Any]: """ - Converts a user type to RECORD that contains n fields, where n is the - number of attributes. Each element in the user type class will be converted to its - corresponding data type in BQ. + Converts a user type to RECORD that contains n fields, where n is the number of attributes. + + Each element in the user type class will be converted to its corresponding data type in BQ. """ names = value._fields values = [self.convert_value(getattr(value, name)) for name in names] @@ -305,17 +302,20 @@ def convert_user_type(self, value: Any) -> dict[str, Any]: def convert_tuple_type(self, values: tuple[Any]) -> dict[str, Any]: """ - Converts a tuple to RECORD that contains n fields, each will be converted - to its corresponding data type in bq and will be named 'field_', where - index is determined by the order of the tuple elements defined in cassandra. + Converts a tuple to RECORD that contains n fields. + + Each field will be converted to its corresponding data type in bq and + will be named 'field_', where index is determined by the order + of the tuple elements defined in cassandra. """ names = ["field_" + str(i) for i in range(len(values))] return self.generate_data_dict(names, values) def convert_map_type(self, value: OrderedMapSerializedKey) -> list[dict[str, Any]]: """ - Converts a map to a repeated RECORD that contains two fields: 'key' and 'value', - each will be converted to its corresponding data type in BQ. + Converts a map to a repeated RECORD that contains two fields: 'key' and 'value'. + + Each will be converted to its corresponding data type in BQ. """ converted_map = [] for k, v in zip(value.keys(), value.values()): diff --git a/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py b/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py index 88b6d09323708..6e59de074fd80 100644 --- a/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py +++ b/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py @@ -443,8 +443,8 @@ def execute(self, context: Context): def execute_complete(self, context: Context, event: dict[str, Any]): """ Callback for when the trigger fires - returns immediately. - Relies on trigger to throw an exception, otherwise it assumes execution was - successful. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. """ if event["status"] == "error": raise AirflowException(event["message"]) @@ -695,6 +695,7 @@ def _validate_src_fmt_configs( ) -> dict: """ Validates the given src_fmt_configs against a valid configuration for the source format. + Adds the backward compatibility config to the src_fmt_configs. :param source_format: File format to export. diff --git a/airflow/providers/google/cloud/transfers/gcs_to_gcs.py b/airflow/providers/google/cloud/transfers/gcs_to_gcs.py index 2b39df1c6a79c..1c091c3214748 100644 --- a/airflow/providers/google/cloud/transfers/gcs_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/gcs_to_gcs.py @@ -312,10 +312,11 @@ def _ignore_existing_files(self, hook, prefix, **kwargs): def _copy_source_without_wildcard(self, hook, prefix): """ - For source_objects with no wildcard, this operator would first list - all files in source_objects, using provided delimiter if any. Then copy - files from source_objects to destination_object and rename each source - file. + List all files in source_objects, copy files to destination_object, and rename each source file. + + For source_objects with no wildcard, this operator would first list all + files in source_objects, using provided delimiter if any. Then copy files + from source_objects to destination_object and rename each source file. Example 1: diff --git a/airflow/providers/google/cloud/transfers/local_to_gcs.py b/airflow/providers/google/cloud/transfers/local_to_gcs.py index 26e70109bc74d..6a287fa95a79f 100644 --- a/airflow/providers/google/cloud/transfers/local_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/local_to_gcs.py @@ -31,8 +31,7 @@ class LocalFilesystemToGCSOperator(BaseOperator): """ - Uploads a file or list of files to Google Cloud Storage. - Optionally can compress the file for upload. + Uploads a file or list of files to Google Cloud Storage; optionally can compress the file for upload. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/transfers/mssql_to_gcs.py b/airflow/providers/google/cloud/transfers/mssql_to_gcs.py index 152145a8cbd33..d9b728295545a 100644 --- a/airflow/providers/google/cloud/transfers/mssql_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/mssql_to_gcs.py @@ -27,8 +27,8 @@ class MSSQLToGCSOperator(BaseSQLToGCSOperator): - """Copy data from Microsoft SQL Server to Google Cloud Storage - in JSON, CSV or Parquet format. + """ + Copy data from Microsoft SQL Server to Google Cloud Storage in JSON, CSV or Parquet format. :param bit_fields: Sequence of fields names of MSSQL "BIT" data type, to be interpreted in the schema as "BOOLEAN". "BIT" fields that won't @@ -99,8 +99,8 @@ def field_to_bigquery(self, field) -> dict[str, str]: @classmethod def convert_type(cls, value, schema_type, **kwargs): """ - Takes a value from MSSQL, and converts it to a value that's safe for - JSON/Google Cloud Storage/BigQuery. + Take a value from MSSQL and convert it to a value safe for JSON/Google Cloud Storage/BigQuery. + Datetime, Date and Time are converted to ISO formatted strings. """ if isinstance(value, decimal.Decimal): diff --git a/airflow/providers/google/cloud/transfers/mysql_to_gcs.py b/airflow/providers/google/cloud/transfers/mysql_to_gcs.py index 8f9e4dad4f842..cefde7c4ab074 100644 --- a/airflow/providers/google/cloud/transfers/mysql_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/mysql_to_gcs.py @@ -94,8 +94,7 @@ def field_to_bigquery(self, field) -> dict[str, str]: def convert_type(self, value, schema_type: str, **kwargs): """ - Takes a value from MySQLdb, and converts it to a value that's safe for - JSON/Google Cloud Storage/BigQuery. + Take a value from MySQLdb and convert it to a value safe for JSON/Google Cloud Storage/BigQuery. * Datetimes are converted to `str(value)` (`datetime.isoformat(' ')`) strings. diff --git a/airflow/providers/google/cloud/transfers/oracle_to_gcs.py b/airflow/providers/google/cloud/transfers/oracle_to_gcs.py index fcf8458ef9ec8..fc96357d51725 100644 --- a/airflow/providers/google/cloud/transfers/oracle_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/oracle_to_gcs.py @@ -87,8 +87,7 @@ def field_to_bigquery(self, field) -> dict[str, str]: def convert_type(self, value, schema_type, **kwargs): """ - Takes a value from Oracle db, and converts it to a value that's safe for - JSON/Google Cloud Storage/BigQuery. + Take a value from Oracle db and convert it to a value safe for JSON/Google Cloud Storage/BigQuery. * Datetimes are converted to UTC seconds. * Decimals are converted to floats. diff --git a/airflow/providers/google/cloud/transfers/postgres_to_gcs.py b/airflow/providers/google/cloud/transfers/postgres_to_gcs.py index b7637db0a9188..f83b47ea325c9 100644 --- a/airflow/providers/google/cloud/transfers/postgres_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/postgres_to_gcs.py @@ -130,8 +130,8 @@ def field_to_bigquery(self, field) -> dict[str, str]: def convert_type(self, value, schema_type, stringify_dict=True): """ - Takes a value from Postgres, and converts it to a value that's safe for - JSON/Google Cloud Storage/BigQuery. + Take a value from Postgres and convert it to a value safe for JSON/Google Cloud Storage/BigQuery. + Timezone aware Datetime are converted to UTC seconds. Unaware Datetime, Date and Time are converted to ISO formatted strings. Decimals are converted to floats. diff --git a/airflow/providers/google/cloud/transfers/presto_to_gcs.py b/airflow/providers/google/cloud/transfers/presto_to_gcs.py index 1433435c7529f..b9385876de6ea 100644 --- a/airflow/providers/google/cloud/transfers/presto_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/presto_to_gcs.py @@ -86,8 +86,10 @@ def execute(self, *args, **kwargs) -> PrestoResult: def executemany(self, *args, **kwargs): """ - Prepare a database operation (query or command) and then execute it against all parameter - sequences or mappings found in the sequence seq_of_parameters. + Prepare and execute a database operation. + + Prepare a database operation (query or command) and then execute it against + all parameter sequences or mappings found in the sequence seq_of_parameters. """ self.initialized = False self.rows = [] @@ -101,18 +103,16 @@ def peekone(self) -> Any: return element def fetchone(self) -> Any: - """ - Fetch the next row of a query result set, returning a single sequence, or - ``None`` when no more data is available. - """ + """Fetch the next row of a query result set, returning a single sequence, or ``None``.""" if self.rows: return self.rows.pop(0) return self.cursor.fetchone() def fetchmany(self, size=None) -> list: """ - Fetch the next set of rows of a query result, returning a sequence of sequences - (e.g. a list of tuples). An empty sequence is returned when no more rows are available. + Fetch the next set of rows of a query result, returning a sequence of sequences. + + An empty sequence is returned when no more rows are available. """ if size is None: size = self.cursor.arraysize @@ -128,8 +128,7 @@ def fetchmany(self, size=None) -> list: def __next__(self) -> Any: """ - Return the next row from the currently executing SQL statement using the same semantics as - ``.fetchone()``. + Return the next row from the current SQL statement using the same semantics as ``.fetchone()``. A ``StopIteration`` exception is raised when the result set is exhausted. """ diff --git a/airflow/providers/google/cloud/transfers/s3_to_gcs.py b/airflow/providers/google/cloud/transfers/s3_to_gcs.py index 39a2f2a68f56e..b0dbdf90daaef 100644 --- a/airflow/providers/google/cloud/transfers/s3_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/s3_to_gcs.py @@ -35,8 +35,7 @@ class S3ToGCSOperator(S3ListOperator): """ - Synchronizes an S3 key, possibly a prefix, with a Google Cloud Storage - destination path. + Synchronizes an S3 key, possibly a prefix, with a Google Cloud Storage destination path. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/cloud/transfers/sql_to_gcs.py b/airflow/providers/google/cloud/transfers/sql_to_gcs.py index ce283ff866171..e696487f65f02 100644 --- a/airflow/providers/google/cloud/transfers/sql_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/sql_to_gcs.py @@ -371,9 +371,7 @@ def _get_file_mime_type(self): return file_mime_type def _configure_csv_file(self, file_handle, schema): - """Configure a csv writer with the file_handle and write schema - as headers for the new file. - """ + """Configure a csv writer with the file_handle and write schema as headers for the new file.""" csv_writer = csv.writer(file_handle, delimiter=self.field_delimiter) csv_writer.writerow(schema) return csv_writer @@ -438,9 +436,9 @@ def _get_col_type_dict(self): def _write_local_schema_file(self, cursor): """ - Takes a cursor, and writes the BigQuery schema for the results to a - local file system. Schema for database will be read from cursor if - not specified. + Takes a cursor, and writes the BigQuery schema for the results to a local file system. + + Schema for database will be read from cursor if not specified. :return: A dictionary where key is a filename to be used as an object name in GCS, and values are file handles to local files that diff --git a/airflow/providers/google/cloud/transfers/trino_to_gcs.py b/airflow/providers/google/cloud/transfers/trino_to_gcs.py index 60dbecb4b73b9..138f69acf1c0c 100644 --- a/airflow/providers/google/cloud/transfers/trino_to_gcs.py +++ b/airflow/providers/google/cloud/transfers/trino_to_gcs.py @@ -86,8 +86,10 @@ def execute(self, *args, **kwargs) -> TrinoResult: def executemany(self, *args, **kwargs): """ - Prepare a database operation (query or command) and then execute it against all parameter - sequences or mappings found in the sequence seq_of_parameters. + Prepare and execute a database query. + + Prepare a database operation (query or command) and then execute it against + all parameter sequences or mappings found in the sequence seq_of_parameters. """ self.initialized = False self.rows = [] @@ -101,18 +103,16 @@ def peekone(self) -> Any: return element def fetchone(self) -> Any: - """ - Fetch the next row of a query result set, returning a single sequence, or - ``None`` when no more data is available. - """ + """Fetch the next row of a query result set, returning a single sequence, or ``None``.""" if self.rows: return self.rows.pop(0) return self.cursor.fetchone() def fetchmany(self, size=None) -> list: """ - Fetch the next set of rows of a query result, returning a sequence of sequences - (e.g. a list of tuples). An empty sequence is returned when no more rows are available. + Fetch the next set of rows of a query result, returning a sequence of sequences. + + An empty sequence is returned when no more rows are available. """ if size is None: size = self.cursor.arraysize @@ -128,8 +128,7 @@ def fetchmany(self, size=None) -> list: def __next__(self) -> Any: """ - Return the next row from the currently executing SQL statement using the same semantics as - ``.fetchone()``. + Return the next row from the current SQL statement using the same semantics as ``.fetchone()``. A ``StopIteration`` exception is raised when the result set is exhausted. """ diff --git a/airflow/providers/google/cloud/triggers/bigquery.py b/airflow/providers/google/cloud/triggers/bigquery.py index bb3973c338e82..b1c659a845a1d 100644 --- a/airflow/providers/google/cloud/triggers/bigquery.py +++ b/airflow/providers/google/cloud/triggers/bigquery.py @@ -521,8 +521,7 @@ async def _table_exists( self, hook: BigQueryTableAsyncHook, dataset: str, table_id: str, project_id: str ) -> bool: """ - Create client session and make call to BigQueryTableAsyncHook and check for the table in - Google Big Query. + Create session, make call to BigQueryTableAsyncHook, and check for the table in Google Big Query. :param hook: BigQueryTableAsyncHook Hook class :param dataset: The name of the dataset in which to look for the table storage bucket. diff --git a/airflow/providers/google/cloud/triggers/bigquery_dts.py b/airflow/providers/google/cloud/triggers/bigquery_dts.py index 95e6aed39711c..def8b90b66a30 100644 --- a/airflow/providers/google/cloud/triggers/bigquery_dts.py +++ b/airflow/providers/google/cloud/triggers/bigquery_dts.py @@ -27,7 +27,9 @@ class BigQueryDataTransferRunTrigger(BaseTrigger): - """Triggers class to watch the Transfer Run state to define when the job is done. + """ + Triggers class to watch the Transfer Run state to define when the job is done. + :param project_id: The BigQuery project id where the transfer configuration should be :param config_id: ID of the config of the Transfer Run which should be watched. :param run_id: ID of the Transfer Run which should be watched. @@ -35,13 +37,13 @@ class BigQueryDataTransferRunTrigger(BaseTrigger): :param gcp_conn_id: The connection ID used to connect to Google Cloud. :param location: BigQuery Transfer Service location for regional transfers. :param impersonation_chain: Optional service account to impersonate using short-term - credentials, or chained list of accounts required to get the access_token - of the last account in the list, which will be impersonated in the request. - If set as a string, the account must grant the originating account - the Service Account Token Creator IAM role. - If set as a sequence, the identities from the list must grant - Service Account Token Creator IAM role to the directly preceding identity, with first - account from the list granting this role to the originating account (templated). + credentials, or chained list of accounts required to get the access_token + of the last account in the list, which will be impersonated in the request. + If set as a string, the account must grant the originating account + the Service Account Token Creator IAM role. + If set as a sequence, the identities from the list must grant + Service Account Token Creator IAM role to the directly preceding identity, with first + account from the list granting this role to the originating account (templated). """ def __init__( @@ -79,10 +81,7 @@ def serialize(self) -> tuple[str, dict[str, Any]]: ) async def run(self) -> AsyncIterator[TriggerEvent]: - """ - Get Transfer Run status and if it one of the statuses which mean end of the job - then yield TriggerEvent object. - """ + """If the Transfer Run is in a terminal state, then yield TriggerEvent object.""" hook = self._get_async_hook() while True: try: diff --git a/airflow/providers/google/cloud/triggers/cloud_sql.py b/airflow/providers/google/cloud/triggers/cloud_sql.py index 7d2cd5a323c64..e04ada9277fcc 100644 --- a/airflow/providers/google/cloud/triggers/cloud_sql.py +++ b/airflow/providers/google/cloud/triggers/cloud_sql.py @@ -28,6 +28,7 @@ class CloudSQLExportTrigger(BaseTrigger): """ Trigger that periodically polls information from Cloud SQL API to verify job status. + Implementation leverages asynchronous transport. """ diff --git a/airflow/providers/google/cloud/triggers/dataproc.py b/airflow/providers/google/cloud/triggers/dataproc.py index 6f2be85124697..3f94c49965061 100644 --- a/airflow/providers/google/cloud/triggers/dataproc.py +++ b/airflow/providers/google/cloud/triggers/dataproc.py @@ -289,6 +289,7 @@ async def run(self) -> AsyncIterator[TriggerEvent]: class DataprocWorkflowTrigger(DataprocBaseTrigger): """ Trigger that periodically polls information from Dataproc API to verify status. + Implementation leverages asynchronous transport. """ diff --git a/airflow/providers/google/cloud/triggers/gcs.py b/airflow/providers/google/cloud/triggers/gcs.py index 9595cc6ca89d0..8b7f3abe783a2 100644 --- a/airflow/providers/google/cloud/triggers/gcs.py +++ b/airflow/providers/google/cloud/triggers/gcs.py @@ -209,6 +209,7 @@ async def _is_blob_updated_after( class GCSPrefixBlobTrigger(GCSBlobTrigger): """ Looks for objects in bucket matching a prefix. + If none found, sleep for interval and check again. Otherwise, return matches. :param bucket: the bucket in the google cloud storage where the objects are residing. @@ -287,14 +288,10 @@ async def _list_blobs_with_prefix(self, hook: GCSAsyncHook, bucket_name: str, pr class GCSUploadSessionTrigger(GCSPrefixBlobTrigger): """ - Checks for changes in the number of objects at prefix in Google Cloud Storage - bucket and returns Trigger Event if the inactivity period has passed with no - increase in the number of objects. - - :param bucket: The Google Cloud Storage bucket where the objects are. - expected. - :param prefix: The name of the prefix to check in the Google cloud - storage bucket. + Return Trigger Event if the inactivity period has passed with no increase in the number of objects. + + :param bucket: The Google Cloud Storage bucket where the objects are expected. + :param prefix: The name of the prefix to check in the Google cloud storage bucket. :param poke_interval: polling period in seconds to check :param inactivity_period: The total seconds of inactivity to designate an upload session is over. Note, this mechanism is not real time and @@ -354,10 +351,7 @@ def serialize(self) -> tuple[str, dict[str, Any]]: ) async def run(self) -> AsyncIterator[TriggerEvent]: - """ - Simple loop until no change in any new files or deleted in list blob is - found for the inactivity_period. - """ + """Loop until no new files or deleted files in list blob for the inactivity_period.""" try: hook = self._get_async_hook() while True: @@ -373,16 +367,12 @@ async def run(self) -> AsyncIterator[TriggerEvent]: yield TriggerEvent({"status": "error", "message": str(e)}) def _get_time(self) -> datetime: - """ - This is just a wrapper of datetime.datetime.now to simplify mocking in the - unittests. - """ + """This is just a wrapper of datetime.datetime.now to simplify mocking in the unittests.""" return datetime.now() def _is_bucket_updated(self, current_objects: set[str]) -> dict[str, str]: """ - Checks whether new objects have been uploaded and the inactivity_period - has passed and updates the state of the sensor accordingly. + Check whether new objects have been uploaded and the inactivity_period has passed; update the state. :param current_objects: set of object ids in bucket during last check. """ diff --git a/airflow/providers/google/cloud/utils/bigquery.py b/airflow/providers/google/cloud/utils/bigquery.py index e96e10a77030b..76dc9637bef6c 100644 --- a/airflow/providers/google/cloud/utils/bigquery.py +++ b/airflow/providers/google/cloud/utils/bigquery.py @@ -22,6 +22,7 @@ def bq_cast(string_field: str, bq_type: str) -> None | int | float | bool | str: """ Helper method that casts a BigQuery row to the appropriate data types. + This is useful because BigQuery returns all fields as strings. """ if string_field is None: @@ -40,7 +41,8 @@ def bq_cast(string_field: str, bq_type: str) -> None | int | float | bool | str: def convert_job_id(job_id: str | list[str], project_id: str, location: str | None) -> Any: """ - Helper method that converts to path: project_id:location:job_id + Helper method that converts to path: project_id:location:job_id. + :param project_id: Required. The ID of the Google Cloud project where workspace located. :param location: Optional. The ID of the Google Cloud region where workspace located. :param job_id: Required. The ID of the job. diff --git a/airflow/providers/google/cloud/utils/credentials_provider.py b/airflow/providers/google/cloud/utils/credentials_provider.py index a5063cd738dcb..b6d7bcc385d84 100644 --- a/airflow/providers/google/cloud/utils/credentials_provider.py +++ b/airflow/providers/google/cloud/utils/credentials_provider.py @@ -15,10 +15,8 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -""" -This module contains a mechanism for providing temporary -Google Cloud authentication. -""" +"""This module contains a mechanism for providing temporary Google Cloud authentication.""" + from __future__ import annotations import json @@ -52,8 +50,7 @@ def build_gcp_conn( project_id: str | None = None, ) -> str: """ - Builds a uri that can be used as :envvar:`AIRFLOW_CONN_{CONN_ID}` with provided service key, - scopes and project id. + Build a uri that can be used as :envvar:`AIRFLOW_CONN_{CONN_ID}` with provided values. :param key_file_path: Path to service key. :param scopes: Required OAuth scopes. @@ -80,8 +77,10 @@ def provide_gcp_credentials( key_file_dict: dict | None = None, ) -> Generator[None, None, None]: """ - Context manager that provides a Google Cloud credentials for application supporting - `Application Default Credentials (ADC) strategy`__. + Context manager that provides Google Cloud credentials for Application Default Credentials (ADC). + + .. seealso:: + `Application Default Credentials (ADC) strategy`__. It can be used to provide credentials for external programs (e.g. gcloud) that expect authorization file in ``GOOGLE_APPLICATION_CREDENTIALS`` environment variable. @@ -117,9 +116,9 @@ def provide_gcp_connection( project_id: str | None = None, ) -> Generator[None, None, None]: """ - Context manager that provides a temporary value of :envvar:`AIRFLOW_CONN_GOOGLE_CLOUD_DEFAULT` - connection. It build a new connection that includes path to provided service json, - required scopes and project id. + Context manager that provides a temporary value of :envvar:`AIRFLOW_CONN_GOOGLE_CLOUD_DEFAULT` connection. + + It builds a new connection that includes path to provided service json, required scopes and project id. :param key_file_path: Path to file with Google Cloud Service Account .json file. :param scopes: OAuth scopes for the connection @@ -365,8 +364,7 @@ def get_credentials_and_project_id(*args, **kwargs) -> tuple[google.auth.credent def _get_scopes(scopes: str | None = None) -> Sequence[str]: """ - Parse a comma-separated string containing OAuth2 scopes if `scopes` is provided. - Otherwise, default scope will be returned. + Parse a comma-separated string containing OAuth2 scopes if `scopes` is provided; otherwise return default. :param scopes: A comma-separated string containing OAuth2 scopes :return: Returns the scope defined in the connection configuration, or the default scope @@ -378,6 +376,8 @@ def _get_target_principal_and_delegates( impersonation_chain: str | Sequence[str] | None = None, ) -> tuple[str | None, Sequence[str] | None]: """ + Get the target_principal and optional list of delegates from impersonation_chain. + Analyze contents of impersonation_chain and return target_principal (the service account to directly impersonate using short-term credentials, if any) and optional list of delegates required to get the access_token of target_principal. diff --git a/airflow/providers/google/cloud/utils/dataform.py b/airflow/providers/google/cloud/utils/dataform.py index d47b8228a1a08..e183e29a1a759 100644 --- a/airflow/providers/google/cloud/utils/dataform.py +++ b/airflow/providers/google/cloud/utils/dataform.py @@ -45,6 +45,7 @@ def make_initialization_workspace_flow( ) -> tuple: """ Creates flow which simulates the initialization of the default project. + :param project_id: Required. The ID of the Google Cloud project where workspace located. :param region: Required. The ID of the Google Cloud region where workspace located. :param repository_id: Required. The ID of the Dataform repository where workspace located. diff --git a/airflow/providers/google/cloud/utils/field_sanitizer.py b/airflow/providers/google/cloud/utils/field_sanitizer.py index de43e8c894989..ef2150824c183 100644 --- a/airflow/providers/google/cloud/utils/field_sanitizer.py +++ b/airflow/providers/google/cloud/utils/field_sanitizer.py @@ -103,9 +103,7 @@ class GcpFieldSanitizerException(AirflowException): - """Thrown when sanitizer finds unexpected field type in the path - (other than dict or array). - """ + """Thrown when sanitizer finds unexpected field type in the path (other than dict or array).""" class GcpBodyFieldSanitizer(LoggingMixin): diff --git a/airflow/providers/google/cloud/utils/field_validator.py b/airflow/providers/google/cloud/utils/field_validator.py index 499cb0f7e33ca..1e853d44ed881 100644 --- a/airflow/providers/google/cloud/utils/field_validator.py +++ b/airflow/providers/google/cloud/utils/field_validator.py @@ -413,10 +413,10 @@ def _validate_field(self, validation_spec, dictionary_to_validate, parent=None, def validate(self, body_to_validate: dict) -> None: """ - Validates if the body (dictionary) follows specification that the validator was - instantiated with. Raises ValidationSpecificationException or - ValidationFieldException in case of problems with specification or the - body not conforming to the specification respectively. + Validates if the body (dictionary) follows specification that the validator was instantiated with. + + Raises ValidationSpecificationException or ValidationFieldException in case of problems + with specification or the body not conforming to the specification respectively. :param body_to_validate: body that must follow the specification :return: None diff --git a/airflow/providers/google/common/hooks/base_google.py b/airflow/providers/google/common/hooks/base_google.py index 3ee344a03a881..30287c27049fb 100644 --- a/airflow/providers/google/common/hooks/base_google.py +++ b/airflow/providers/google/common/hooks/base_google.py @@ -93,8 +93,7 @@ def is_soft_quota_exception(exception: Exception): def is_operation_in_progress_exception(exception: Exception) -> bool: """ - Some of the calls return 429 (too many requests!) or 409 errors (Conflict) - in case of operation in progress. + Some calls return 429 (too many requests!) or 409 errors (Conflict) in case of operation in progress. * Google Cloud SQL """ @@ -142,11 +141,12 @@ def get_field(extras: dict, field_name: str): class GoogleBaseHook(BaseHook): """ - A base hook for Google cloud-related hooks. Google cloud has a shared REST - API client that is built in the same way no matter which service you use. - This class helps construct and authorize the credentials needed to then - call googleapiclient.discovery.build() to actually discover and build a client - for a Google cloud service. + A base hook for Google cloud-related hooks. + + Google cloud has a shared REST API client that is built in the same way no matter + which service you use. This class helps construct and authorize the credentials + needed to then call googleapiclient.discovery.build() to actually discover and + build a client for a Google cloud service. The class also contains some miscellaneous helper functions. @@ -326,10 +326,7 @@ def _get_credentials_email(self) -> str: return oauth2_client.tokeninfo().execute()["email"] def _authorize(self) -> google_auth_httplib2.AuthorizedHttp: - """ - Returns an authorized HTTP object to be used to build a Google cloud - service hook connection. - """ + """Returns an authorized HTTP object to be used to build a Google cloud service hook connection.""" credentials = self.get_credentials() http = build_http() http = set_user_agent(http, "airflow/" + version.version) @@ -338,10 +335,11 @@ def _authorize(self) -> google_auth_httplib2.AuthorizedHttp: def _get_field(self, f: str, default: Any = None) -> Any: """ - Fetches a field from extras, and returns it. This is some Airflow - magic. The google_cloud_platform hook type adds custom UI elements - to the hook page, which allow admins to specify service_account, - key_path, etc. They get formatted as shown below. + Fetches a field from extras, and returns it. + + This is some Airflow magic. The google_cloud_platform hook type adds + custom UI elements to the hook page, which allow admins to specify + service_account, key_path, etc. They get formatted as shown below. """ return hasattr(self, "extras") and get_field(self.extras, f) or default @@ -407,10 +405,7 @@ def scopes(self) -> Sequence[str]: @staticmethod def quota_retry(*args, **kwargs) -> Callable: - """ - A decorator that provides a mechanism to repeat requests in response to exceeding a temporary quote - limit. - """ + """Provides a mechanism to repeat requests in response to exceeding a temporary quota limit.""" def decorator(fun: Callable): default_kwargs = { @@ -426,11 +421,7 @@ def decorator(fun: Callable): @staticmethod def operation_in_progress_retry(*args, **kwargs) -> Callable[[T], T]: - """ - A decorator that provides a mechanism to repeat requests in response to - operation in progress (HTTP 409) - limit. - """ + """Provides a mechanism to repeat requests in response to operation in progress (HTTP 409) limit.""" def decorator(fun: T): default_kwargs = { @@ -447,8 +438,9 @@ def decorator(fun: T): @staticmethod def fallback_to_default_project_id(func: Callable[..., RT]) -> Callable[..., RT]: """ - Decorator that provides fallback for Google Cloud project id. If - the project is None it will be replaced with the project_id from the + Decorator that provides fallback for Google Cloud project id. + + If the project is None it will be replaced with the project_id from the service account the Hook is authenticated with. Project id can be specified either via project_id kwarg or via first parameter in positional args. @@ -479,12 +471,11 @@ def inner_wrapper(self: GoogleBaseHook, *args, **kwargs) -> RT: @staticmethod def provide_gcp_credential_file(func: T) -> T: """ - Function decorator that provides a Google Cloud credentials for application supporting Application - Default Credentials (ADC) strategy. + Provides a Google Cloud credentials for Application Default Credentials (ADC) strategy support. - It is recommended to use ``provide_gcp_credential_file_as_context`` context manager to limit the - scope when authorization data is available. Using context manager also - makes it easier to use multiple connection in one function. + It is recommended to use ``provide_gcp_credential_file_as_context`` context + manager to limit the scope when authorization data is available. Using context + manager also makes it easier to use multiple connection in one function. """ @functools.wraps(func) @@ -497,8 +488,11 @@ def wrapper(self: GoogleBaseHook, *args, **kwargs): @contextmanager def provide_gcp_credential_file_as_context(self) -> Generator[str | None, None, None]: """ - Context manager that provides a Google Cloud credentials for application supporting `Application - Default Credentials (ADC) strategy `__. + Provides a Google Cloud credentials for Application Default Credentials (ADC) strategy support. + + See: + `Application Default Credentials (ADC) + strategy `__. It can be used to provide credentials for external programs (e.g. gcloud) that expect authorization file in ``GOOGLE_APPLICATION_CREDENTIALS`` environment variable. @@ -588,10 +582,10 @@ def provide_authorized_gcloud(self) -> Generator[None, None, None]: def download_content_from_request(file_handle, request: dict, chunk_size: int) -> None: """ Download media resources. - Note that the Python file object is compatible with io.Base and can be used with this class also. - :param file_handle: io.Base or file object. The stream in which to write the downloaded - bytes. + Note that the Python file object is compatible with io.Base and can be used with this class also. + + :param file_handle: io.Base or file object. The stream in which to write the downloaded bytes. :param request: googleapiclient.http.HttpRequest, the media request to perform in chunks. :param chunk_size: int, File will be downloaded in chunks of this many bytes. """ @@ -628,10 +622,7 @@ def __init__(self, **kwargs: Any): self._sync_hook = None async def get_sync_hook(self) -> Any: - """ - Sync version of the Google Cloud Hooks makes blocking calls in ``__init__`` so we don't inherit - from it. - """ + """Sync version of the Google Cloud Hook makes blocking calls in ``__init__``; don't inherit it.""" if not self._sync_hook: self._sync_hook = await sync_to_async(self.sync_hook_class)(**self._hook_kwargs) return self._sync_hook diff --git a/airflow/providers/google/common/hooks/discovery_api.py b/airflow/providers/google/common/hooks/discovery_api.py index 13daf7ccf7ce6..da0b9f9ce4946 100644 --- a/airflow/providers/google/common/hooks/discovery_api.py +++ b/airflow/providers/google/common/hooks/discovery_api.py @@ -82,8 +82,7 @@ def get_conn(self) -> Resource: def query(self, endpoint: str, data: dict, paginate: bool = False, num_retries: int = 0) -> dict: """ - Creates a dynamic API call to any Google API registered in Google's API Client Library - and queries it. + Creates a dynamic API call to any Google API registered in Google's API Client Library and queries it. :param endpoint: The client libraries path to the api call's executing method. For example: 'analyticsreporting.reports.batchGet' diff --git a/airflow/providers/google/firebase/hooks/firestore.py b/airflow/providers/google/firebase/hooks/firestore.py index 93fc154e997af..58c9811829734 100644 --- a/airflow/providers/google/firebase/hooks/firestore.py +++ b/airflow/providers/google/firebase/hooks/firestore.py @@ -110,8 +110,7 @@ def export_documents( def _wait_for_operation_to_complete(self, operation_name: str) -> None: """ - Waits for the named operation to complete - checks status of the - asynchronous call. + Waits for the named operation to complete - checks status of the asynchronous call. :param operation_name: The name of the operation. :return: The response returned by the operation. diff --git a/airflow/providers/google/firebase/operators/firestore.py b/airflow/providers/google/firebase/operators/firestore.py index af7ec700bc319..d14242e6b0461 100644 --- a/airflow/providers/google/firebase/operators/firestore.py +++ b/airflow/providers/google/firebase/operators/firestore.py @@ -28,8 +28,7 @@ class CloudFirestoreExportDatabaseOperator(BaseOperator): """ - Exports a copy of all or a subset of documents from Google Cloud Firestore to another storage system, - such as Google Cloud Storage. + Export documents from Google Cloud Firestore to another storage system, such as Google Cloud Storage. .. seealso:: For more information on how to use this operator, take a look at the guide: diff --git a/airflow/providers/google/marketing_platform/operators/analytics.py b/airflow/providers/google/marketing_platform/operators/analytics.py index 2996838f12c2f..0098980e9bee2 100644 --- a/airflow/providers/google/marketing_platform/operators/analytics.py +++ b/airflow/providers/google/marketing_platform/operators/analytics.py @@ -385,8 +385,9 @@ def execute(self, context: Context) -> None: class GoogleAnalyticsModifyFileHeadersDataImportOperator(BaseOperator): """ - GA has a very particular naming convention for Data Import. Ability to - prefix "ga:" to all column headers and also a dict to rename columns to + GA has a very particular naming convention for Data Import. + + Ability to prefix "ga:" to all column headers and also a dict to rename columns to match the custom dimension ID in GA i.e clientId : dimensionX. :param storage_bucket: The Google cloud storage bucket where the file is stored. diff --git a/airflow/providers/google/suite/transfers/gcs_to_gdrive.py b/airflow/providers/google/suite/transfers/gcs_to_gdrive.py index c1e796258f534..9220b6b3b59d9 100644 --- a/airflow/providers/google/suite/transfers/gcs_to_gdrive.py +++ b/airflow/providers/google/suite/transfers/gcs_to_gdrive.py @@ -35,8 +35,7 @@ class GCSToGoogleDriveOperator(BaseOperator): """ - Copies objects from a Google Cloud Storage service to a Google Drive service, with renaming - if requested. + Copies objects from a Google Cloud Storage service to a Google Drive service, with renaming if requested. Using this operator requires the following OAuth 2.0 scope: