Move template-field validation/transformation out of operator __init__ (exemption-list burn-down)

### Background

Template fields are rendered **after** an operator's constructor runs. Any logic applied to a template-field parameter's **value** inside `__init__` — validation, type checks, transformation, string interpolation — therefore operates on the un-rendered Jinja expression, not the real value.

This is documented in [Creating a custom Operator](https://airflow.apache.org/docs/apache-airflow/stable/howto/custom-operator.html#templating) and [contributing-docs/05_pull_requests.rst](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst), and it causes real bugs: #69813 (`SSHRemoteJobOperator` validated a templated `remote_base_dir` in `__init__`, so cleanup validation failed for any custom base) is a recent example.

> [!IMPORTANT]
> Checks that only ask **whether an argument was passed** are the exception and belong in `__init__`. Read [False positives](#false-positives) below before fixing an entry.

The `validate-operators-init` prek hook previously only verified that template fields are *assigned* verbatim at the top level of `__init__`; it could not see validation calls, conditionals, or transformations (and it did not cover sensors, `AwsBaseOperator[...]` subclasses, or `aws_template_fields(...)`-based classes at all). The hook has been extended to detect any non-sanctioned read of a template field in `__init__` (PR to follow).

### The exemption ratchet

Existing violations are listed in [`scripts/ci/prek/validate_operators_init_exemptions.txt`](https://github.com/apache/airflow/blob/main/scripts/ci/prek/validate_operators_init_exemptions.txt) as `path::ClassName` entries so the hook can enforce the rule on new code immediately.

- A PR that fixes a class **must remove its entry** in the same PR — the hook fails on stale exemptions, so this cannot be forgotten.
- New violations cannot be added: only files listed in the exemption file are suppressed.
- This issue is done when the exemption file is empty.

**How to fix a class:** move the validation / transformation from `__init__` into `execute()` (or the first method that runs after rendering), keep `__init__` down to plain `self.field = field` assignments (defaulting via `field or default` is fine). Don't forget the corresponding tests.

Please do **not** open sub-issues for individual entries — comment here or just open a PR referencing this issue.

### False positives

Not everything the hook flags should move. A check that only asks **whether an argument was passed** — the usual "exactly one of `a` or `b`" guard — belongs in `__init__` and must not be moved:

- The constructor is the only place that can answer it. With `render_template_as_native_obj=True` a *provided* field renders to `None`, so the same check in `execute()` reports a supplied argument as missing.
- Raising at construction surfaces a static authoring mistake as a Dag import error, instead of on a worker once per task instance and per retry.

**Fix these by rewriting in place, not by moving.** Use the `is not None` polarity — `if field:` is a truthiness test on the un-rendered string and asks a third question that matches neither:

```python
# in __init__ — correct
if not exactly_one(command is not None, powershell is not None, cmdlet is not None):
raise ValueError("Must provide exactly one of 'command', 'powershell', or 'cmdlet'")
```

Do **not** write `exactly_one(command is None, ...)`. It happens to agree for two arguments but is wrong for three or more: it rejects the valid single-argument case and accepts two arguments at once.

Anything that inspects the *value* — a range check, a format check, an `in` test against allowed values, `.strip()`, `isinstance(...)` — still moves to `execute()`. And a provision check in `__init__` guarantees nothing about the rendered value: code in `execute()` that needs the field set still needs its own guard.

The hook did not know this distinction until it was narrowed. These five entries were flagged only for a provision check and clear with **no operator code change** — they are removed from the exemption file by the narrowing PR, so please don't pick them up:

| class | file | the flagged check |
| --- | --- | --- |
| `EmrAddStepsOperator` | [emr.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py) | `exactly_one(job_flow_id is None, job_flow_name is None)` |
| `GCSDeleteObjectsOperator` | [gcs.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/gcs.py) | `objects` / `prefix` exclusivity |
| `GCSToLocalFilesystemOperator` | [gcs_to_local.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_local.py) | `filename` / `store_to_xcom_key` exclusivity |
| `OracleToAzureDataLakeOperator` | [oracle_to_azure_data_lake.py](https://github.com/apache/airflow/blob/main/providers/microsoft/azure/src/airflow/providers/microsoft/azure/transfers/oracle_to_azure_data_lake.py) | `sql_params` default |
| `OracleToOracleOperator` | [oracle_to_oracle.py](https://github.com/apache/airflow/blob/main/providers/oracle/src/airflow/providers/oracle/transfers/oracle_to_oracle.py) | `source_sql_params` default |

Five already-merged fixes moved a provision check that should have stayed. Nothing is broken, but those files now demonstrate the discouraged pattern — reverting them is tracked in #70503. Please don't use them as a model.

`S3DeleteObjectsOperator` needs care and is worth reading before you touch it. Its check is a pure provision check, but it hides the comparisons in a comprehension (`all(var is None for var in [...])`) that the hook does not recognise, so the entry stays even after the narrowing. It also **already exists twice** — in `__init__` and in `execute()` — and both copies must stay:

- The `execute()` copy is load-bearing. `keys = self.keys or self.hook.list_keys(prefix=..., from_datetime=..., to_datetime=...)` lists the **whole bucket** when every filter is `None`, and the next line deletes it. A templated `keys` that renders to `None` (native rendering) reaches that state past a correct `__init__` check.
- The `__init__` copy still catches the static authoring mistake at Dag parse time rather than on a worker.

To clear the entry, unroll the `__init__` copy so every read is a direct comparison — don't delete it, and don't touch `execute()`:

```python
by_scan = prefix is not None or from_datetime is not None or to_datetime is not None
if not exactly_one(keys is not None, by_scan):
raise ValueError("Either keys or at least one of prefix, from_datetime, to_datetime should be set.")
```

That is semantically identical to the current condition on every input combination (including `keys=[]`) and clears the hook.

### Current exemption list (snapshot)

The authoritative list is the exemptions file; this checklist is the snapshot at the time the check was introduced. **83 classes** across 22 providers, of which the 5 listed under [False positives](#false-positives) above need no code change.

**amazon**

- [x] `AppflowBaseOperator` — [appflow.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/appflow.py) — logic in `__init__`
- [x] `AwsToAwsBaseOperator` — [base.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/transfers/base.py) — logic in `__init__`
- [x] `BedrockCreateKnowledgeBaseOperator` — [bedrock.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/bedrock.py) — logic in `__init__`
- [x] `BedrockRaGOperator` — [bedrock.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/bedrock.py) — transformed assignment, logic in `__init__`, missing assignment
- [x] `DataSyncOperator` — [datasync.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/datasync.py) — logic in `__init__`
- [x] `DmsModifyTaskOperator` — [dms.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/dms.py) — logic in `__init__`
- [x] `DmsStartReplicationOperator` — [dms.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/dms.py) — logic in `__init__`
- [x] `EcsRunTaskOperator` — [ecs.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/ecs.py) — logic in `__init__`
- [x] `GCSToS3Operator` — [gcs_to_s3.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/transfers/gcs_to_s3.py) — logic in `__init__` **in PR #71591**
- [x] `GlueDataQualityOperator` — [glue.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py) — transformed assignment, logic in `__init__`, missing assignment
- [x] `MongoToS3Operator` — [mongo_to_s3.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/transfers/mongo_to_s3.py) — logic in `__init__`
- [ ] `NeptuneStartDbClusterOperator` — [neptune.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py) — transformed assignment, missing assignment **in PR #70491**
- [ ] `NeptuneStopDbClusterOperator` — [neptune.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py) — transformed assignment, missing assignment **in PR #70491**
- [x] `S3DeleteObjectsOperator` — [s3.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/s3.py) — logic in `__init__`
- [x] `S3ToRedshiftOperator` — [s3_to_redshift.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/transfers/s3_to_redshift.py) — logic in `__init__`
- [x] `SageMakerCreateNotebookOperator` — [sagemaker.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/sagemaker.py) — logic in `__init__`
- [x] `SageMakerProcessingOperator` — [sagemaker.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/sagemaker.py) — logic in `__init__`
- [x] `StepFunctionStartExecutionOperator` — [step_function.py](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/step_function.py) — transformed assignment, missing assignment

**anthropic**

- [x] `AnthropicAgentSessionOperator` — [agent.py](https://github.com/apache/airflow/blob/main/providers/anthropic/src/airflow/providers/anthropic/operators/agent.py) — logic in `__init__`

**apache/hive**

- [x] `HivePartitionSensor` — [hive_partition.py](https://github.com/apache/airflow/blob/main/providers/apache/hive/src/airflow/providers/apache/hive/sensors/hive_partition.py) — logic in `__init__`
- [x] `NamedHivePartitionSensor` — [named_hive_partition.py](https://github.com/apache/airflow/blob/main/providers/apache/hive/src/airflow/providers/apache/hive/sensors/named_hive_partition.py) — logic in `__init__`

**apache/kafka**

- [x] `ProduceToTopicOperator` — [produce.py](https://github.com/apache/airflow/blob/main/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py) — logic in `__init__`

**apache/spark**

- [x] `SparkSubmitOperator` — [spark_submit.py](https://github.com/apache/airflow/blob/main/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py) — logic in `__init__`

**cncf/kubernetes**

- [x] `KubernetesInstallKueueOperator` — [kueue.py](https://github.com/apache/airflow/blob/main/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/kueue.py) — logic in `__init__`
- [ ] `KubernetesPodOperator` — [pod.py](https://github.com/apache/airflow/blob/main/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py) — logic in `__init__`
- [x] `KubernetesResourceBaseOperator` — [resource.py](https://github.com/apache/airflow/blob/main/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/resource.py) — logic in `__init__`

**cohere**

- [x] `CohereEmbeddingOperator` — [embedding.py](https://github.com/apache/airflow/blob/main/providers/cohere/src/airflow/providers/cohere/operators/embedding.py) — logic in `__init__`

**common/ai**

- [x] `AgentOperator` — [agent.py](https://github.com/apache/airflow/blob/main/providers/common/ai/src/airflow/providers/common/ai/operators/agent.py) — logic in `__init__`
- [x] `DocumentLoaderOperator` — [document_loader.py](https://github.com/apache/airflow/blob/main/providers/common/ai/src/airflow/providers/common/ai/operators/document_loader.py) — logic in `__init__`

**databricks**

- [x] `DatabricksCopyIntoOperator` — [databricks_sql.py](https://github.com/apache/airflow/blob/main/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py) — logic in `__init__`
- [x] `DatabricksReposCreateOperator` — [databricks_repos.py](https://github.com/apache/airflow/blob/main/providers/databricks/src/airflow/providers/databricks/operators/databricks_repos.py) — logic in `__init__`
- [x] `DatabricksReposDeleteOperator` — [databricks_repos.py](https://github.com/apache/airflow/blob/main/providers/databricks/src/airflow/providers/databricks/operators/databricks_repos.py) — logic in `__init__`
- [x] `DatabricksReposUpdateOperator` — [databricks_repos.py](https://github.com/apache/airflow/blob/main/providers/databricks/src/airflow/providers/databricks/operators/databricks_repos.py) — logic in `__init__`
- [x] `DatabricksSQLStatementsSensor` — [databricks.py](https://github.com/apache/airflow/blob/main/providers/databricks/src/airflow/providers/databricks/sensors/databricks.py) — logic in `__init__`

**dbt/cloud**

- [x] `DbtCloudGetJobRunArtifactOperator` — [dbt.py](https://github.com/apache/airflow/blob/main/providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py) — logic in `__init__`

**docker**

- [x] `DockerOperator` — [docker.py](https://github.com/apache/airflow/blob/main/providers/docker/src/airflow/providers/docker/operators/docker.py) — logic in `__init__`

**google**

- [x] `AzureFileShareToGCSOperator` — [azure_fileshare_to_gcs.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py) — logic in `__init__` **in PR #70740**
- [ ] `BigQueryDataTransferServiceTransferRunSensor` — [bigquery_dts.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/sensors/bigquery_dts.py) — transformed assignment, logic in `__init__`, missing assignment **in PR #70528**
- [x] `BigQueryInsertJobOperator` — [bigquery.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py) — logic in `__init__`
- [x] `BigQueryToMsSqlOperator` — [bigquery_to_mssql.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/transfers/bigquery_to_mssql.py) — logic in `__init__`
- [x] `CloudBatchSubmitJobOperator` — [cloud_batch.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/cloud_batch.py) — logic in `__init__`
- [ ] `CloudBuildCreateBuildOperator` — [cloud_build.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/cloud_build.py) — logic in `__init__`
- [ ] `CloudComposerExternalTaskSensor` — [cloud_composer.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/sensors/cloud_composer.py) — logic in `__init__`
- [ ] `CloudDataTransferServiceCreateJobOperator` — [cloud_storage_transfer_service.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py) — logic in `__init__` **in PR #70529**
- [ ] `CloudFunctionDeployFunctionOperator` — [functions.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/functions.py) — logic in `__init__` **in PR #70531**
- [x] `ComputeEngineCopyInstanceTemplateOperator` — [compute.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/compute.py) — logic in `__init__`
- [x] `ComputeEngineDeleteInstanceGroupManagerOperator` — [compute.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/compute.py) — logic in `__init__`
- [x] `ComputeEngineDeleteInstanceOperator` — [compute.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/compute.py) — logic in `__init__`
- [x] `ComputeEngineDeleteInstanceTemplateOperator` — [compute.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/compute.py) — logic in `__init__`
- [x] `ComputeEngineInsertInstanceFromTemplateOperator` — [compute.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/compute.py) — logic in `__init__`
- [x] `ComputeEngineInsertInstanceGroupManagerOperator` — [compute.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/compute.py) — logic in `__init__`
- [x] `ComputeEngineInsertInstanceOperator` — [compute.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/compute.py) — logic in `__init__`
- [x] `ComputeEngineInsertInstanceTemplateOperator` — [compute.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/compute.py) — logic in `__init__`
- [x] `ComputeEngineInstanceGroupUpdateManagerTemplateOperator` — [compute.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/compute.py) — logic in `__init__`
- [x] `ComputeEngineSetMachineTypeOperator` — [compute.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/compute.py) — logic in `__init__`
- [ ] `DataprocCreateClusterOperator` — [dataproc.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/dataproc.py) — logic in `__init__`
- [x] ~`DataprocSubmitJobOperator` — [dataproc.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/dataproc.py) — logic in `__init__`~ False positive (see https://github.com/apache/airflow/pull/73040)
- [ ] `GCSFileTransformOperator` — [gcs.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/gcs.py) — logic in `__init__` **in PR #70488**
- [x] `GCSListObjectsOperator` — [gcs.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/gcs.py) — logic in `__init__`
- [x] `GCSToBigQueryOperator` — [gcs_to_bigquery.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py) — logic in `__init__` **in PR #70542**
- [x] `GCSToGCSOperator` — [gcs_to_gcs.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py) — logic in `__init__` **in PR #70449**
- [x] `GenAIGeminiCreateBatchJobOperator` — [gen_ai.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/gen_ai.py) — logic in `__init__`
- [x] `GenAIGeminiCreateEmbeddingsBatchJobOperator` — [gen_ai.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/cloud/operators/gen_ai.py) — logic in `__init__`
- [x] `GoogleCampaignManagerDeleteReportOperator` — [campaign_manager.py](https://github.com/apache/airflow/blob/main/providers/google/src/airflow/providers/google/marketing_platform/operators/campaign_manager.py) — logic in `__init__` **in PR #70530**

**microsoft/azure**

- [x] `AzureVirtualMachineStateSensor` — [compute.py](https://github.com/apache/airflow/blob/main/providers/microsoft/azure/src/airflow/providers/microsoft/azure/sensors/compute.py) — logic in `__init__`
- [x] `GCSToAzureBlobStorageOperator` — [gcs_to_wasb.py](https://github.com/apache/airflow/blob/main/providers/microsoft/azure/src/airflow/providers/microsoft/azure/transfers/gcs_to_wasb.py) — logic in `__init__` **in PR #70574**

**microsoft/psrp**

- [x] `PsrpOperator` — [psrp.py](https://github.com/apache/airflow/blob/main/providers/microsoft/psrp/src/airflow/providers/microsoft/psrp/operators/psrp.py) — logic in `__init__` **in PR #70347**

**neo4j**

- [x] `Neo4jOperator` — [neo4j.py](https://github.com/apache/airflow/blob/main/providers/neo4j/src/airflow/providers/neo4j/operators/neo4j.py) — logic in `__init__`

**papermill**

- [x] `PapermillOperator` — [papermill.py](https://github.com/apache/airflow/blob/main/providers/papermill/src/airflow/providers/papermill/operators/papermill.py) — logic in `__init__`

**snowflake**

- [x] `SnowparkContainerJobOperator` — [snowpark_containers.py](https://github.com/apache/airflow/blob/main/providers/snowflake/src/airflow/providers/snowflake/operators/snowpark_containers.py) — logic in `__init__`

**ssh**

- [x] `SSHOperator` — [ssh.py](https://github.com/apache/airflow/blob/main/providers/ssh/src/airflow/providers/ssh/operators/ssh.py) — logic in `__init__`
- [x] `SSHRemoteJobOperator` — [ssh_remote_job.py](https://github.com/apache/airflow/blob/main/providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py) — logic in `__init__`

**standard**

- [x] `BashOperator` — [bash.py](https://github.com/apache/airflow/blob/main/providers/standard/src/airflow/providers/standard/operators/bash.py) — logic in `__init__`
- [x] `DateTimeSensor` — [date_time.py](https://github.com/apache/airflow/blob/main/providers/standard/src/airflow/providers/standard/sensors/date_time.py) — logic in `__init__`, missing assignment
- [x] `HITLOperator` — [hitl.py](https://github.com/apache/airflow/blob/main/providers/standard/src/airflow/providers/standard/operators/hitl.py) — logic in `__init__`
- [x] `TriggerDagRunOperator` — [trigger_dagrun.py](https://github.com/apache/airflow/blob/main/providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py) — logic in `__init__`

**teradata**

- [x] `TeradataToTeradataOperator` — [teradata_to_teradata.py](https://github.com/apache/airflow/blob/main/providers/teradata/src/airflow/providers/teradata/transfers/teradata_to_teradata.py) — logic in `__init__`

**weaviate**

- [x] `WeaviateIngestOperator` — [weaviate.py](https://github.com/apache/airflow/blob/main/providers/weaviate/src/airflow/providers/weaviate/operators/weaviate.py) — logic in `__init__`

---

Drafted-by: Claude Code (Opus 4.8)

Start with scripts/ci/prek/validate_operators_init_exemptions.txt and the validate-operators-init hook, then choose an unclaimed provider class named in the checklist and read its referenced operator file and tests. Move only value validation or transformation out of __init__ as described, preserve provision checks, run the relevant tests and hook, and verify the exemption entry is removed.

Written by the indexing model from the issue text.