Add CloudSQLNoOperationInProgressSensor for parallel admin ops

providers/google/docs/operators/cloud/cloud_sql.rst

@@ -359,6 +359,45 @@ as shown in the example:
     :start-after: [START howto_operator_cloudsql_import_gcs_permissions]
     :end-before: [END howto_operator_cloudsql_import_gcs_permissions]
 
+.. _howto/operator:CloudSQLNoOperationInProgressSensor:
+
+CloudSQLNoOperationInProgressSensor
+-----------------------------------
+
+Cloud SQL serializes administrative operations per instance: only one import, export, backup or
+similar operation can run against an instance at a time. Submitting another while one is in flight
+fails immediately with HTTP 409 ``operationInProgress``. This commonly affects DAGs that fan out
+multiple :class:`~airflow.providers.google.cloud.operators.cloud_sql.CloudSQLImportInstanceOperator`
+/ :class:`~airflow.providers.google.cloud.operators.cloud_sql.CloudSQLExportInstanceOperator` tasks
+against the same instance in parallel.
+
+Use
+:class:`~airflow.providers.google.cloud.sensors.cloud_sql.CloudSQLNoOperationInProgressSensor`
+to wait until the instance has no administrative operation in progress before submitting the next
+one. The sensor polls ``sqladmin.operations.list`` for the instance and succeeds once no operation
+is in a non-terminal (``PENDING`` / ``RUNNING``) state. It supports deferrable mode and fails fast
+on HTTP 403/404 (the instance is missing or access is denied).
+
+.. code-block:: python
+
+    from airflow.providers.google.cloud.sensors.cloud_sql import (
+        CloudSQLNoOperationInProgressSensor,
+    )
+
+    wait_for_slot = CloudSQLNoOperationInProgressSensor(
+        task_id="wait_for_slot",
+        instance="my-cloudsql-pg",
+        poke_interval=60,
+        timeout=2 * 60 * 60,
+        deferrable=True,
+    )
+
+    wait_for_slot >> import_data
+
+The sensor is best-effort: it reduces the chance of a 409 but cannot guarantee exclusivity, since a
+new operation (for example one triggered from the console or by an automated backup) could start
+between the sensor passing and the operator submitting.
+
 .. _howto/operator:CloudSQLCreateInstanceOperator:
 
 CloudSQLCreateInstanceOperator

providers/google/provider.yaml

@@ -650,6 +650,9 @@ sensors:
   - integration-name: Google Bigtable
     python-modules:
       - airflow.providers.google.cloud.sensors.bigtable
+  - integration-name: Google Cloud SQL
+    python-modules:
+      - airflow.providers.google.cloud.sensors.cloud_sql
   - integration-name: Managed Service for Apache Airflow
     python-modules:
       - airflow.providers.google.cloud.sensors.cloud_composer

providers/google/src/airflow/providers/google/cloud/hooks/cloud_sql.py

@@ -92,6 +92,15 @@ class CloudSqlOperationStatus:
     UNKNOWN = "UNKNOWN"
 
 
+# Statuses that mean an administrative operation is still in flight on the instance. Cloud SQL
+# serializes admin operations per instance, so a new import/export submitted while one of these is
+# active fails with HTTP 409 ``operationInProgress``. Keying off an explicit set (rather than
+# ``status != DONE``) avoids treating UNKNOWN/unexpected statuses as in-progress and poking forever.
+CLOUD_SQL_NON_TERMINAL_STATUSES = frozenset(
+    {CloudSqlOperationStatus.PENDING, CloudSqlOperationStatus.RUNNING}
+)
+
+
 class CloudSQLHook(GoogleBaseHook):
     """
     Hook for Google Cloud SQL APIs.
@@ -429,6 +438,29 @@ def get_operation(self, project_id: str, operation_name: str) -> dict:
             .execute(num_retries=self.num_retries)
         )
 
+    @GoogleBaseHook.fallback_to_default_project_id
+    def list_operations(self, instance: str, project_id: str, max_results: int | None = None) -> list[dict]:
+        """
+        List administrative operations for a Cloud SQL instance.
+
+        Must be called with keyword arguments because ``project_id`` is injected by the
+        ``fallback_to_default_project_id`` decorator.
+
+        :param instance: Name of the Cloud SQL instance whose operations are listed.
+        :param project_id: Project ID of the project that contains the instance.
+        :param max_results: Optional maximum number of operations to return per page.
+        :return: The list of operation resources for the instance (may be empty).
+        """
+        response = (
+            self.get_conn()
+            .operations()
+            .list(project=project_id, instance=instance, maxResults=max_results)
+            .execute(num_retries=self.num_retries)
+        )
+        # ``operations.list`` already filters server-side by ``instance``; keep a defensive
+        # client-side filter on ``targetId`` in case the API ever returns broader results.
+        return [op for op in response.get("items", []) if op.get("targetId") == instance]
+
     @GoogleBaseHook.fallback_to_default_project_id
     def _wait_for_operation_to_complete(
         self, project_id: str, operation_name: str, time_to_sleep: int = TIME_TO_SLEEP_IN_SECONDS

providers/google/src/airflow/providers/google/cloud/sensors/cloud_sql.py

@@ -0,0 +1,147 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""This module contains Google Cloud SQL sensors."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from datetime import timedelta
+from typing import TYPE_CHECKING
+
+from googleapiclient.errors import HttpError
+
+from airflow.providers.common.compat.sdk import AirflowException, BaseSensorOperator, conf
+from airflow.providers.google.cloud.hooks.cloud_sql import (
+    CLOUD_SQL_NON_TERMINAL_STATUSES,
+    CloudSQLHook,
+)
+from airflow.providers.google.cloud.triggers.cloud_sql import CloudSQLNoOperationInProgressTrigger
+from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID
+
+if TYPE_CHECKING:
+    from airflow.providers.common.compat.sdk import Context
+
+
+class CloudSQLOperationError(AirflowException):
+    """Raised when the Cloud SQL operations check fails (for example, the instance is missing or access is denied)."""
+
+
+class CloudSQLNoOperationInProgressSensor(BaseSensorOperator):
+    """
+    Wait until a Cloud SQL instance has no administrative operation in progress.
+
+    Cloud SQL serializes administrative operations per instance: only one import, export, backup or
+    similar operation can run at a time. Submitting another while one is in flight fails with HTTP
+    409 ``operationInProgress``. Place this sensor upstream of
+    :class:`~airflow.providers.google.cloud.operators.cloud_sql.CloudSQLImportInstanceOperator` /
+    :class:`~airflow.providers.google.cloud.operators.cloud_sql.CloudSQLExportInstanceOperator`
+    (or between mutually exclusive admin operators) to serialize work against the same instance.
+
+    The sensor is operation-agnostic: it polls ``sqladmin.operations.list`` for the instance and
+    succeeds once no operation is in a non-terminal (PENDING/RUNNING) state. It is best-effort -- a
+    new operation could still start between the sensor passing and the next operator submitting.
+
+    .. seealso::
+        For more information on how to use this sensor, take a look at the guide:
+        :ref:`howto/operator:CloudSQLNoOperationInProgressSensor`
+
+    :param instance: Name of the Cloud SQL instance to watch.
+    :param project_id: Optional, Google Cloud Project ID. If not provided the default project is used.
+    :param gcp_conn_id: The connection ID used to connect to Google Cloud.
+    :param api_version: API version used (e.g. v1beta4).
+    :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.
+    :param deferrable: Run the sensor in deferrable mode.
+    """
+
+    template_fields: Sequence[str] = (
+        "project_id",
+        "instance",
+        "impersonation_chain",
+    )
+    ui_color = "#D4ECEA"
+
+    def __init__(
+        self,
+        *,
+        instance: str,
+        project_id: str = PROVIDE_PROJECT_ID,
+        gcp_conn_id: str = "google_cloud_default",
+        api_version: str = "v1beta4",
+        impersonation_chain: str | Sequence[str] | None = None,
+        deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self.instance = instance
+        self.project_id = project_id
+        self.gcp_conn_id = gcp_conn_id
+        self.api_version = api_version
+        self.impersonation_chain = impersonation_chain
+        self.deferrable = deferrable
+
+    def _get_hook(self) -> CloudSQLHook:
+        return CloudSQLHook(
+            api_version=self.api_version,
+            gcp_conn_id=self.gcp_conn_id,
+            impersonation_chain=self.impersonation_chain,
+        )
+
+    def poke(self, context: Context) -> bool:
+        hook = self._get_hook()
+        try:
+            operations = hook.list_operations(project_id=self.project_id, instance=self.instance)
+        except HttpError as e:
+            if e.resp.status in (403, 404):
+                # Instance missing or access denied - surface the misconfiguration instead of poking.
+                raise CloudSQLOperationError(
+                    f"Cloud SQL operations.list failed for instance {self.instance}: {e}"
+                )
+            raise
+        in_progress = [op for op in operations if op.get("status") in CLOUD_SQL_NON_TERMINAL_STATUSES]
+        if in_progress:
+            self.log.info(
+                "%s operation(s) still in progress on instance %s.", len(in_progress), self.instance
+            )
+            return False
+        return True
+
+    def execute(self, context: Context) -> None:
+        """Run on the worker and defer using the trigger when in deferrable mode."""
+        if self.deferrable:
+            if not self.poke(context=context):
+                self.defer(
+                    timeout=timedelta(seconds=self.timeout),
+                    trigger=CloudSQLNoOperationInProgressTrigger(
+                        instance=self.instance,
+                        project_id=self.project_id,
+                        gcp_conn_id=self.gcp_conn_id,
+                        impersonation_chain=self.impersonation_chain,
+                        poke_interval=int(self.poke_interval),
+                        api_version=self.api_version,
+                    ),
+                    method_name="execute_complete",
+                )
+        else:
+            super().execute(context)
+
+    def execute_complete(self, context: Context, event: dict | None = None) -> None:
+        """Act as a callback for when the trigger fires."""
+        if event and event.get("status") in ("failed", "error"):
+            raise CloudSQLOperationError(event["message"])
+        self.log.info("No administrative operation in progress on instance %s.", self.instance)

providers/google/src/airflow/providers/google/cloud/triggers/cloud_sql.py

@@ -23,8 +23,13 @@
 from collections.abc import Sequence
 
 from asgiref.sync import sync_to_async
+from googleapiclient.errors import HttpError
 
-from airflow.providers.google.cloud.hooks.cloud_sql import CloudSQLAsyncHook, CloudSqlOperationStatus
+from airflow.providers.google.cloud.hooks.cloud_sql import (
+    CLOUD_SQL_NON_TERMINAL_STATUSES,
+    CloudSQLAsyncHook,
+    CloudSqlOperationStatus,
+)
 from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 
@@ -117,3 +122,77 @@ async def run(self):
                     "message": str(e),
                 }
             )
+
+
+class CloudSQLNoOperationInProgressTrigger(BaseTrigger):
+    """
+    Trigger that waits until a Cloud SQL instance has no administrative operation in progress.
+
+    Polls ``sqladmin.operations.list`` for the target instance and fires once no operation is in a
+    non-terminal state (PENDING/RUNNING). Fails fast on 403/404 (the instance is missing or access
+    is denied) rather than polling until timeout.
+    """
+
+    def __init__(
+        self,
+        instance: str,
+        project_id: str = PROVIDE_PROJECT_ID,
+        gcp_conn_id: str = "google_cloud_default",
+        impersonation_chain: str | Sequence[str] | None = None,
+        poke_interval: int = 20,
+        api_version: str = "v1beta4",
+    ):
+        super().__init__()
+        self.instance = instance
+        self.project_id = project_id
+        self.gcp_conn_id = gcp_conn_id
+        self.impersonation_chain = impersonation_chain
+        self.poke_interval = poke_interval
+        self.api_version = api_version
+        self.hook = CloudSQLAsyncHook(
+            gcp_conn_id=self.gcp_conn_id,
+            impersonation_chain=self.impersonation_chain,
+        )
+
+    def serialize(self):
+        return (
+            "airflow.providers.google.cloud.triggers.cloud_sql.CloudSQLNoOperationInProgressTrigger",
+            {
+                "instance": self.instance,
+                "project_id": self.project_id,
+                "gcp_conn_id": self.gcp_conn_id,
+                "impersonation_chain": self.impersonation_chain,
+                "poke_interval": self.poke_interval,
+                "api_version": self.api_version,
+            },
+        )
+
+    async def run(self):
+        try:
+            sync_hook = await self.hook.get_sync_hook(api_version=self.api_version)
+            while True:
+                # No async ``operations.list`` exists on the hook, so run the sync call in a thread.
+                operations = await sync_to_async(sync_hook.list_operations)(
+                    project_id=self.project_id, instance=self.instance
+                )
+                in_progress = [op for op in operations if op.get("status") in CLOUD_SQL_NON_TERMINAL_STATUSES]
+                if not in_progress:
+                    yield TriggerEvent({"instance": self.instance, "status": "success"})
+                    return
+                self.log.info(
+                    "%s operation(s) still in progress on instance %s, sleeping for %s seconds.",
+                    len(in_progress),
+                    self.instance,
+                    self.poke_interval,
+                )
+                await asyncio.sleep(self.poke_interval)
+        except HttpError as e:
+            if e.resp.status in (403, 404):
+                # Instance missing or access denied - no point retrying.
+                yield TriggerEvent({"status": "failed", "message": str(e)})
+                return
+            self.log.exception("Error listing operations for instance %s.", self.instance)
+            yield TriggerEvent({"status": "failed", "message": str(e)})
+        except Exception as e:
+            self.log.exception("Error listing operations for instance %s.", self.instance)
+            yield TriggerEvent({"status": "failed", "message": str(e)})

providers/google/src/airflow/providers/google/get_provider_info.py

@@ -720,6 +720,10 @@ def get_provider_info():
                 "integration-name": "Google Bigtable",
                 "python-modules": ["airflow.providers.google.cloud.sensors.bigtable"],
             },
+            {
+                "integration-name": "Google Cloud SQL",
+                "python-modules": ["airflow.providers.google.cloud.sensors.cloud_sql"],
+            },
             {
                 "integration-name": "Managed Service for Apache Airflow",
                 "python-modules": ["airflow.providers.google.cloud.sensors.cloud_composer"],

providers/google/tests/system/google/cloud/cloud_sql/example_cloud_sql.py

@@ -46,6 +46,7 @@
     GCSDeleteBucketOperator,
     GCSObjectCreateAclEntryOperator,
 )
+from airflow.providers.google.cloud.sensors.cloud_sql import CloudSQLNoOperationInProgressSensor
 
 try:
     from airflow.sdk import TriggerRule
@@ -228,6 +229,15 @@
     )
     # [END howto_operator_cloudsql_import_gcs_permissions]
 
+    # Cloud SQL serializes admin operations per instance, so wait until the export above has
+    # finished (no operation in progress) before submitting the import to avoid a 409.
+    # [START howto_sensor_cloudsql_no_operation_in_progress]
+    sql_wait_no_operation_task = CloudSQLNoOperationInProgressSensor(
+        instance=INSTANCE_NAME,
+        task_id="sql_wait_no_operation_task",
+    )
+    # [END howto_sensor_cloudsql_no_operation_in_progress]
+
     # [START howto_operator_cloudsql_import]
     sql_import_task = CloudSQLImportInstanceOperator(
         body=import_body, instance=INSTANCE_NAME, task_id="sql_import_task"
@@ -288,6 +298,7 @@
         >> sql_export_task
         >> sql_export_def_task
         >> sql_gcp_add_object_permission_task
+        >> sql_wait_no_operation_task
         >> sql_import_task
         >> sql_instance_clone
         >> sql_db_delete_task