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

[shahar1]

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 and 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 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 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:

# 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	exactly_one(job_flow_id is None, job_flow_name is None)
GCSDeleteObjectsOperator	gcs.py	objects / prefix exclusivity
GCSToLocalFilesystemOperator	gcs_to_local.py	filename / store_to_xcom_key exclusivity
OracleToAzureDataLakeOperator	oracle_to_azure_data_lake.py	sql_params default
OracleToOracleOperator	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():

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 above need no code change.

amazon

• AppflowBaseOperator — appflow.py — logic in __init__
• AwsToAwsBaseOperator — base.py — logic in __init__
• BedrockCreateKnowledgeBaseOperator — bedrock.py — logic in __init__
• BedrockRaGOperator — bedrock.py — transformed assignment, logic in __init__, missing assignment
• DataSyncOperator — datasync.py — logic in __init__
• DmsModifyTaskOperator — dms.py — logic in __init__
• DmsStartReplicationOperator — dms.py — logic in __init__
• EcsRunTaskOperator — ecs.py — logic in __init__
• GCSToS3Operator — gcs_to_s3.py — logic in __init__ in PR Check GCSToS3Operator match_glob support after template rendering #70723
• GlueDataQualityOperator — glue.py — transformed assignment, logic in __init__, missing assignment
• MongoToS3Operator — mongo_to_s3.py — logic in __init__
• NeptuneStartDbClusterOperator — neptune.py — transformed assignment, missing assignment in PR Rename Neptune operator db_cluster_id argument to cluster_id #70491
• NeptuneStopDbClusterOperator — neptune.py — transformed assignment, missing assignment in PR Rename Neptune operator db_cluster_id argument to cluster_id #70491
• S3DeleteObjectsOperator — s3.py — logic in __init__
• S3ToRedshiftOperator — s3_to_redshift.py — logic in __init__
• SageMakerCreateNotebookOperator — sagemaker.py — logic in __init__
• SageMakerProcessingOperator — sagemaker.py — logic in __init__
• StepFunctionStartExecutionOperator — step_function.py — transformed assignment, missing assignment

anthropic

• AnthropicAgentSessionOperator — agent.py — logic in __init__

apache/hive

• HivePartitionSensor — hive_partition.py — logic in __init__
• NamedHivePartitionSensor — named_hive_partition.py — logic in __init__

apache/kafka

• ProduceToTopicOperator — produce.py — logic in __init__

apache/spark

• SparkSubmitOperator — spark_submit.py — logic in __init__

cncf/kubernetes

• KubernetesInstallKueueOperator — kueue.py — logic in __init__
• KubernetesPodOperator — pod.py — logic in __init__
• KubernetesResourceBaseOperator — resource.py — logic in __init__

cohere

• CohereEmbeddingOperator — embedding.py — logic in __init__

common/ai

• AgentOperator — agent.py — logic in __init__
• DocumentLoaderOperator — document_loader.py — logic in __init__

databricks

• DatabricksCopyIntoOperator — databricks_sql.py — logic in __init__
• DatabricksReposCreateOperator — databricks_repos.py — logic in __init__
• DatabricksReposDeleteOperator — databricks_repos.py — logic in __init__
• DatabricksReposUpdateOperator — databricks_repos.py — logic in __init__
• DatabricksSQLStatementsSensor — databricks.py — logic in __init__

dbt/cloud

• DbtCloudGetJobRunArtifactOperator — dbt.py — logic in __init__

docker

• DockerOperator — docker.py — logic in __init__

google

• AzureFileShareToGCSOperator — azure_fileshare_to_gcs.py — logic in __init__ in PR Resolve AzureFileShareToGCSOperator directory_name alias after rendering #70740
• BigQueryDataTransferServiceTransferRunSensor — bigquery_dts.py — transformed assignment, logic in __init__, missing assignment in PR Normalize BigQuery DTS sensor expected statuses after rendering #70528
• BigQueryInsertJobOperator — bigquery.py — logic in __init__
• BigQueryToMsSqlOperator — bigquery_to_mssql.py — logic in __init__
• CloudBatchSubmitJobOperator — cloud_batch.py — logic in __init__
• CloudBuildCreateBuildOperator — cloud_build.py — logic in __init__
• CloudComposerExternalTaskSensor — cloud_composer.py — logic in __init__
• CloudDataTransferServiceCreateJobOperator — cloud_storage_transfer_service.py — logic in __init__ in PR Validate storage transfer job body after template rendering #70529
• CloudFunctionDeployFunctionOperator — functions.py — logic in __init__ in PR Validate Cloud Function deploy body after template rendering #70531
• ComputeEngineCopyInstanceTemplateOperator — compute.py — logic in __init__
• ComputeEngineDeleteInstanceGroupManagerOperator — compute.py — logic in __init__
• ComputeEngineDeleteInstanceOperator — compute.py — logic in __init__
• ComputeEngineDeleteInstanceTemplateOperator — compute.py — logic in __init__
• ComputeEngineInsertInstanceFromTemplateOperator — compute.py — logic in __init__
• ComputeEngineInsertInstanceGroupManagerOperator — compute.py — logic in __init__
• ComputeEngineInsertInstanceOperator — compute.py — logic in __init__
• ComputeEngineInsertInstanceTemplateOperator — compute.py — logic in __init__
• ComputeEngineInstanceGroupUpdateManagerTemplateOperator — compute.py — logic in __init__
• ComputeEngineSetMachineTypeOperator — compute.py — logic in __init__
• DataprocCreateClusterOperator — dataproc.py — logic in __init__
• DataprocSubmitJobOperator — dataproc.py — logic in __init__
• GCSFileTransformOperator — gcs.py — logic in __init__ in PR Resolve GCSFileTransformOperator destination fallbacks after rendering #70488
• GCSListObjectsOperator — gcs.py — logic in __init__
• GCSToBigQueryOperator — gcs_to_bigquery.py — logic in __init__ in PR Emit GCSToBigQueryOperator deprecation warning after rendering #70542
• GCSToGCSOperator — gcs_to_gcs.py — logic in __init__ in PR Emit GCSToGCSOperator deprecation warnings after rendering #70449
• GenAIGeminiCreateBatchJobOperator — gen_ai.py — logic in __init__
• GenAIGeminiCreateEmbeddingsBatchJobOperator — gen_ai.py — logic in __init__
• GoogleCampaignManagerDeleteReportOperator — campaign_manager.py — logic in __init__ in PR Keep Campaign Manager delete report provision check in __init__ #70530

microsoft/azure

• AzureVirtualMachineStateSensor — compute.py — logic in __init__
• GCSToAzureBlobStorageOperator — gcs_to_wasb.py — logic in __init__ in PR Check GCSToAzureBlobStorageOperator match_glob support after template… #70574

microsoft/psrp

• PsrpOperator — psrp.py — logic in __init__ in PR Fix PsrpOperator option checks and drop the cmdlet task_id default #70347

neo4j

• Neo4jOperator — neo4j.py — logic in __init__

papermill

• PapermillOperator — papermill.py — logic in __init__

snowflake

• SnowparkContainerJobOperator — snowpark_containers.py — logic in __init__

ssh

• SSHOperator — ssh.py — logic in __init__
• SSHRemoteJobOperator — ssh_remote_job.py — logic in __init__

standard

• BashOperator — bash.py — logic in __init__
• DateTimeSensor — date_time.py — logic in __init__, missing assignment
• HITLOperator — hitl.py — logic in __init__
• TriggerDagRunOperator — trigger_dagrun.py — logic in __init__

teradata

• TeradataToTeradataOperator — teradata_to_teradata.py — logic in __init__

weaviate

• WeaviateIngestOperator — weaviate.py — logic in __init__

---

Drafted-by: Claude Code (Opus 4.8) (no human review before posting)

[justinpakzad]

I will work on the update to SnowparkContainerJobOperator.

[Andrushika]

Hi @shahar1, could we mark the completed ones in the table? (maybe add a checkbox in the list)
I think it’d make it easier to see what’s still left.

[FrankYang0529]

I will work on ProduceToTopicOperator. Thank you.

#70351

[shahar1]

Hi @shahar1, could we mark the completed ones in the table? (maybe add a checkbox in the list) I think it’d make it easier to see what’s still left.

Yes, I'll handle it later

[yuseok89]

I'd like to take the GCS operators on the Google side, starting with GCSDeleteObjectsOperator, GCSFileTransformOperator, and GCSListObjectsOperator.

[Avcu]

I would like to work on BedrockCreateKnowledgeBaseOperator and BedrockRaGOperator on the Amazon side.

[Andrushika]

I would like to take k8s part: KubernetesInstallKueueOperator and KubernetesResourceBaseOperator, thanks! Will wait until #70297 is merged before opening PRs.

[shahar1]

Thanks everyone for your help!
I'd like to recall:

1. There will be no official assignment by current policy - just create the PRs in your time (feel free to tag me once in the PR's description/first comment).
2. While competing PRs are not forbidden, please try to avoid stepping on each other's feet and be patient before recreating an existing PR - there are plenty of operators. Notifying in this GitHub issue is recommended for coordination.

[SZL741023]

I would like to take the AgentOperator and DocumentLoaderOperator thx.

[ahilashsasidharan]

I would like to take on the anthropic AnthropicAgentSessionOperator