Add CloudSQLNoOperationInProgressSensor for parallel admin ops

[Vamsi-klu]

What

Adds an operation-agnostic, deferrable CloudSQLNoOperationInProgressSensor to the Google provider. The sensor waits until a Cloud SQL instance has no administrative operation in flight before downstream tasks submit a new one.

Supporting changes:

• Add CloudSQLHook.list_operations() for sqladmin.operations.list, filtered to the requested instance via targetId.
• Add CLOUD_SQL_NON_TERMINAL_STATUSES (PENDING/RUNNING) to decide whether an operation is still in progress.
• Add CloudSQLNoOperationInProgressTrigger for deferrable mode.
• Register the sensor module in provider.yaml, generated provider info, the Cloud SQL how-to docs, and the system-test example.

Why

Cloud SQL serializes administrative operations per instance. Two parallel CloudSQLImportInstanceOperator / CloudSQLExportInstanceOperator tasks against the same instance can race, and the second submit can fail immediately with HTTP 409 operationInProgress.

This sensor reduces that collision window by giving Dags an explicit pre-flight wait before an admin operation is submitted.

Relationship to #68361

This PR is complementary to the hook-level retry work in #68361, not a replacement for it.

• Fix Cloud SQL 409 operationInProgress on import/export operations #68361 is the submit-path correctness fix: it can retry a 409 that still happens after an operator submits.
• This PR is an orchestration primitive: it lets a Dag wait before submit, and in deferrable mode it does not hold a worker slot while waiting.

The sensor remains best-effort. It cannot eliminate the race between sensor success and the next operator submit, but it gives users a reusable way to reduce avoidable collisions and to serialize mixed Cloud SQL admin operations before they occupy worker execution.

Testing

New/updated unit tests:

• providers/google/tests/unit/google/cloud/hooks/test_cloud_sql.py::TestGcpSqlHookDefaultProjectId::test_list_operations
• providers/google/tests/unit/google/cloud/sensors/test_cloud_sql.py
• providers/google/tests/unit/google/cloud/triggers/test_cloud_sql.py::TestCloudSQLNoOperationInProgressTrigger

Validation run after rebasing onto current upstream/main:

• uv run ruff format providers/google/src/airflow/providers/google/cloud/hooks/cloud_sql.py providers/google/src/airflow/providers/google/cloud/sensors/cloud_sql.py providers/google/src/airflow/providers/google/cloud/triggers/cloud_sql.py providers/google/src/airflow/providers/google/get_provider_info.py providers/google/tests/system/google/cloud/cloud_sql/example_cloud_sql.py providers/google/tests/unit/google/cloud/hooks/test_cloud_sql.py providers/google/tests/unit/google/cloud/sensors/test_cloud_sql.py providers/google/tests/unit/google/cloud/triggers/test_cloud_sql.py
• uv run ruff check --fix providers/google/src/airflow/providers/google/cloud/hooks/cloud_sql.py providers/google/src/airflow/providers/google/cloud/sensors/cloud_sql.py providers/google/src/airflow/providers/google/cloud/triggers/cloud_sql.py providers/google/src/airflow/providers/google/get_provider_info.py providers/google/tests/system/google/cloud/cloud_sql/example_cloud_sql.py providers/google/tests/unit/google/cloud/hooks/test_cloud_sql.py providers/google/tests/unit/google/cloud/sensors/test_cloud_sql.py providers/google/tests/unit/google/cloud/triggers/test_cloud_sql.py
• uv run --project providers/google pytest providers/google/tests/unit/google/cloud/hooks/test_cloud_sql.py::TestGcpSqlHookDefaultProjectId::test_list_operations providers/google/tests/unit/google/cloud/sensors/test_cloud_sql.py providers/google/tests/unit/google/cloud/triggers/test_cloud_sql.py::TestCloudSQLNoOperationInProgressTrigger -xvs
• prek run ruff --from-ref upstream/main
• prek run ruff-format --from-ref upstream/main

Live GCP E2E was not run in this environment.

Related to: #68040

---

Was generative AI tooling used to co-author this PR?

• Yes - Codex (GPT-5); Claude Code (GPT-5)

Generated-by: Codex (GPT-5); Claude Code (GPT-5) following the guidelines

[potiuk]

@Vamsi-klu A few things need addressing before review — see our Pull Request quality criteria.

• ❌ Merge conflicts with main. See docs.

No rush.

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[potiuk]

Thanks — approving this as the surviving implementation of CloudSQLNoOperationInProgressSensor.

Context worth recording: #68146 proposes the same sensor and trigger, opened a day before this one. Two contributors built the same primitive in parallel and both sat for ~8 weeks, which is the worst outcome for everyone. I'm going with this one and closing that.

What decided it:

• This is mergeable; #68146 is conflicting and ~1042 commits behind.
• CLOUD_SQL_NON_TERMINAL_STATUSES is factored into the hook and shared, rather than the status list being duplicated across sensor and trigger.
• CloudSQLOperationError(AirflowException) is a dedicated exception subclass — exactly what the project asks for instead of raising the broad AirflowException directly. There are no new bare raise AirflowException(...) statements here.
• Trigger tests live in test_cloud_sql.py, mirroring triggers/cloud_sql.py, per the test-location convention.

The sensor being operation-agnostic (waiting on any non-terminal admin operation rather than a specific one) is the right shape — it's the 409 operationInProgress case that Dag authors actually hit, and it composes with import/export/clone/patch without needing a variant per operation type.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting