From 9834b600bfea506d319f7915ea59db120db1bcf9 Mon Sep 17 00:00:00 2001
From: AlejandroMorgante <62363051+AlejandroMorgante@users.noreply.github.com>
Date: Tue, 14 Jul 2026 19:27:59 -0300
Subject: [PATCH 1/2] Add Amazon ECR repository operators

Container-based AWS workflows need reusable repository setup and cleanup. AgentCore deployments share this lifecycle, and the SageMaker system tests currently implement it with direct SDK calls.
---
 providers/amazon/docs/operators/ecr.rst       |  87 +++++++
 providers/amazon/provider.yaml                |   5 +
 .../providers/amazon/aws/operators/ecr.py     | 233 ++++++++++++++++++
 .../providers/amazon/get_provider_info.py     |   5 +
 .../tests/system/amazon/aws/example_ecr.py    |  99 ++++++++
 .../unit/amazon/aws/operators/test_ecr.py     | 208 ++++++++++++++++
 6 files changed, 637 insertions(+)
 create mode 100644 providers/amazon/docs/operators/ecr.rst
 create mode 100644 providers/amazon/src/airflow/providers/amazon/aws/operators/ecr.py
 create mode 100644 providers/amazon/tests/system/amazon/aws/example_ecr.py
 create mode 100644 providers/amazon/tests/unit/amazon/aws/operators/test_ecr.py

diff --git a/providers/amazon/docs/operators/ecr.rst b/providers/amazon/docs/operators/ecr.rst
new file mode 100644
index 0000000000000..6c35efb869c28
--- /dev/null
+++ b/providers/amazon/docs/operators/ecr.rst
@@ -0,0 +1,87 @@
+ .. 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.
+
+=======================================
+Amazon Elastic Container Registry (ECR)
+=======================================
+
+`Amazon Elastic Container Registry (ECR) <https://aws.amazon.com/ecr/>`__ is a managed container registry
+for storing, sharing, and deploying container images and artifacts.
+
+Prerequisite Tasks
+------------------
+
+.. include:: ../_partials/prerequisite_tasks.rst
+
+Generic Parameters
+------------------
+
+.. include:: ../_partials/generic_parameters.rst
+
+Operators
+---------
+
+.. _howto/operator:EcrCreateRepositoryOperator:
+
+Create an ECR repository
+========================
+
+To create an ECR repository, use
+:class:`~airflow.providers.amazon.aws.operators.ecr.EcrCreateRepositoryOperator`.
+The operator returns the complete response from the Boto3 ``create_repository`` API operation.
+
+.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_ecr.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_operator_ecr_create_repository]
+    :end-before: [END howto_operator_ecr_create_repository]
+
+.. _howto/operator:EcrSetRepositoryPolicyOperator:
+
+Set an ECR repository policy
+============================
+
+To set the repository policy for an ECR repository, use
+:class:`~airflow.providers.amazon.aws.operators.ecr.EcrSetRepositoryPolicyOperator`.
+The operator returns the complete response from the Boto3 ``set_repository_policy`` API operation.
+
+.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_ecr.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_operator_ecr_set_repository_policy]
+    :end-before: [END howto_operator_ecr_set_repository_policy]
+
+.. _howto/operator:EcrDeleteRepositoryOperator:
+
+Delete an ECR repository
+========================
+
+To delete an ECR repository, use
+:class:`~airflow.providers.amazon.aws.operators.ecr.EcrDeleteRepositoryOperator`.
+Set ``force=True`` to delete a repository that contains images. The operator returns the complete response
+from the Boto3 ``delete_repository`` API operation.
+
+.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_ecr.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_operator_ecr_delete_repository]
+    :end-before: [END howto_operator_ecr_delete_repository]
+
+Reference
+---------
+
+* `AWS Boto3 library documentation for ECR <https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecr.html>`__
diff --git a/providers/amazon/provider.yaml b/providers/amazon/provider.yaml
index 4fa06f211212d..afc766814820a 100644
--- a/providers/amazon/provider.yaml
+++ b/providers/amazon/provider.yaml
@@ -182,6 +182,8 @@ integrations:
   - integration-name: Amazon Elastic Container Registry (ECR)
     external-doc-url: https://aws.amazon.com/ecr/
     logo: /docs/integration-logos/Amazon-Elastic-Container-Registry_light-bg@4x.png
+    how-to-guide:
+      - /docs/apache-airflow-providers-amazon/operators/ecr.rst
     tags: [aws]
   - integration-name: Amazon ECS
     external-doc-url: https://aws.amazon.com/ecs/
@@ -447,6 +449,9 @@ operators:
   - integration-name: Amazon EC2
     python-modules:
       - airflow.providers.amazon.aws.operators.ec2
+  - integration-name: Amazon Elastic Container Registry (ECR)
+    python-modules:
+      - airflow.providers.amazon.aws.operators.ecr
   - integration-name: Amazon ECS
     python-modules:
       - airflow.providers.amazon.aws.operators.ecs
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/ecr.py b/providers/amazon/src/airflow/providers/amazon/aws/operators/ecr.py
new file mode 100644
index 0000000000000..738fafde86ad5
--- /dev/null
+++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/ecr.py
@@ -0,0 +1,233 @@
+# 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.
+"""Amazon Elastic Container Registry (ECR) operators."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import TYPE_CHECKING, Any, ClassVar
+
+from airflow.providers.amazon.aws.hooks.ecr import EcrHook
+from airflow.providers.amazon.aws.operators.base_aws import AwsBaseOperator
+from airflow.providers.amazon.aws.utils.mixins import aws_template_fields
+from airflow.utils.helpers import prune_dict
+
+if TYPE_CHECKING:
+    from airflow.sdk import Context
+
+
+class EcrCreateRepositoryOperator(AwsBaseOperator[EcrHook]):
+    """
+    Create an Amazon ECR repository.
+
+    .. seealso::
+        For more information on how to use this operator, take a look at the guide:
+        :ref:`howto/operator:EcrCreateRepositoryOperator`
+
+    :param repository_name: The name of the repository to create. (templated)
+    :param registry_id: The AWS account ID associated with the registry. (templated)
+    :param tags: Metadata to apply to the repository. (templated)
+    :param image_tag_mutability: The tag mutability setting for the repository. (templated)
+    :param image_tag_mutability_exclusion_filters: Filters that override the repository's
+        image tag mutability setting. (templated)
+    :param image_scanning_configuration: The image scanning configuration for the repository. (templated)
+    :param encryption_configuration: The encryption configuration for the repository. (templated)
+    :param aws_conn_id: The Airflow connection used for AWS credentials.
+        If this is ``None`` or empty then the default boto3 behaviour is used. If
+        running Airflow in a distributed manner and aws_conn_id is None or
+        empty, then default boto3 configuration would be used (and must be
+        maintained on each worker node).
+    :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+    :param verify: Whether or not to verify SSL certificates. See:
+        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+    :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+        https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+    """
+
+    aws_hook_class = EcrHook
+    template_fields: Sequence[str] = aws_template_fields(
+        "repository_name",
+        "registry_id",
+        "tags",
+        "image_tag_mutability",
+        "image_tag_mutability_exclusion_filters",
+        "image_scanning_configuration",
+        "encryption_configuration",
+    )
+    template_fields_renderers: ClassVar[dict[str, str]] = {
+        "tags": "json",
+        "image_tag_mutability_exclusion_filters": "json",
+        "image_scanning_configuration": "json",
+        "encryption_configuration": "json",
+    }
+
+    def __init__(
+        self,
+        *,
+        repository_name: str,
+        registry_id: str | None = None,
+        tags: list[dict[str, str]] | None = None,
+        image_tag_mutability: str | None = None,
+        image_tag_mutability_exclusion_filters: list[dict[str, str]] | None = None,
+        image_scanning_configuration: dict[str, bool] | None = None,
+        encryption_configuration: dict[str, str] | None = None,
+        aws_conn_id: str | None = "aws_default",
+        **kwargs,
+    ):
+        super().__init__(aws_conn_id=aws_conn_id, **kwargs)
+        self.repository_name = repository_name
+        self.registry_id = registry_id
+        self.tags = tags
+        self.image_tag_mutability = image_tag_mutability
+        self.image_tag_mutability_exclusion_filters = image_tag_mutability_exclusion_filters
+        self.image_scanning_configuration = image_scanning_configuration
+        self.encryption_configuration = encryption_configuration
+
+    def execute(self, context: Context) -> dict[str, Any]:
+        self.log.info("Creating Amazon ECR repository %s", self.repository_name)
+        response = self.hook.conn.create_repository(
+            **prune_dict(
+                {
+                    "registryId": self.registry_id,
+                    "repositoryName": self.repository_name,
+                    "tags": self.tags,
+                    "imageTagMutability": self.image_tag_mutability,
+                    "imageTagMutabilityExclusionFilters": self.image_tag_mutability_exclusion_filters,
+                    "imageScanningConfiguration": self.image_scanning_configuration,
+                    "encryptionConfiguration": self.encryption_configuration,
+                }
+            )
+        )
+        self.log.info("Created Amazon ECR repository %s", self.repository_name)
+        return response
+
+
+class EcrSetRepositoryPolicyOperator(AwsBaseOperator[EcrHook]):
+    """
+    Set the repository policy for an Amazon ECR repository.
+
+    .. seealso::
+        For more information on how to use this operator, take a look at the guide:
+        :ref:`howto/operator:EcrSetRepositoryPolicyOperator`
+
+    :param repository_name: The name of the repository to receive the policy. (templated)
+    :param policy_text: The JSON repository policy text to apply. (templated)
+    :param registry_id: The AWS account ID associated with the registry. (templated)
+    :param force: Whether to replace an existing policy that prevents setting a new policy. (templated)
+    :param aws_conn_id: The Airflow connection used for AWS credentials.
+        If this is ``None`` or empty then the default boto3 behaviour is used. If
+        running Airflow in a distributed manner and aws_conn_id is None or
+        empty, then default boto3 configuration would be used (and must be
+        maintained on each worker node).
+    :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+    :param verify: Whether or not to verify SSL certificates. See:
+        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+    :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+        https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+    """
+
+    aws_hook_class = EcrHook
+    template_fields: Sequence[str] = aws_template_fields(
+        "repository_name", "policy_text", "registry_id", "force"
+    )
+    template_fields_renderers: ClassVar[dict[str, str]] = {"policy_text": "json"}
+
+    def __init__(
+        self,
+        *,
+        repository_name: str,
+        policy_text: str,
+        registry_id: str | None = None,
+        force: bool = False,
+        aws_conn_id: str | None = "aws_default",
+        **kwargs,
+    ):
+        super().__init__(aws_conn_id=aws_conn_id, **kwargs)
+        self.repository_name = repository_name
+        self.policy_text = policy_text
+        self.registry_id = registry_id
+        self.force = force
+
+    def execute(self, context: Context) -> dict[str, Any]:
+        self.log.info("Setting repository policy for Amazon ECR repository %s", self.repository_name)
+        response = self.hook.conn.set_repository_policy(
+            **prune_dict(
+                {
+                    "registryId": self.registry_id,
+                    "repositoryName": self.repository_name,
+                    "policyText": self.policy_text,
+                    "force": self.force,
+                }
+            )
+        )
+        self.log.info("Set repository policy for Amazon ECR repository %s", self.repository_name)
+        return response
+
+
+class EcrDeleteRepositoryOperator(AwsBaseOperator[EcrHook]):
+    """
+    Delete an Amazon ECR repository.
+
+    .. seealso::
+        For more information on how to use this operator, take a look at the guide:
+        :ref:`howto/operator:EcrDeleteRepositoryOperator`
+
+    :param repository_name: The name of the repository to delete. (templated)
+    :param registry_id: The AWS account ID associated with the registry. (templated)
+    :param force: Whether to delete the repository when it contains images. (templated)
+    :param aws_conn_id: The Airflow connection used for AWS credentials.
+        If this is ``None`` or empty then the default boto3 behaviour is used. If
+        running Airflow in a distributed manner and aws_conn_id is None or
+        empty, then default boto3 configuration would be used (and must be
+        maintained on each worker node).
+    :param region_name: AWS region_name. If not specified then the default boto3 behaviour is used.
+    :param verify: Whether or not to verify SSL certificates. See:
+        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+    :param botocore_config: Configuration dictionary (key-values) for botocore client. See:
+        https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+    """
+
+    aws_hook_class = EcrHook
+    template_fields: Sequence[str] = aws_template_fields("repository_name", "registry_id", "force")
+
+    def __init__(
+        self,
+        *,
+        repository_name: str,
+        registry_id: str | None = None,
+        force: bool = False,
+        aws_conn_id: str | None = "aws_default",
+        **kwargs,
+    ):
+        super().__init__(aws_conn_id=aws_conn_id, **kwargs)
+        self.repository_name = repository_name
+        self.registry_id = registry_id
+        self.force = force
+
+    def execute(self, context: Context) -> dict[str, Any]:
+        self.log.info("Deleting Amazon ECR repository %s", self.repository_name)
+        response = self.hook.conn.delete_repository(
+            **prune_dict(
+                {
+                    "registryId": self.registry_id,
+                    "repositoryName": self.repository_name,
+                    "force": self.force,
+                }
+            )
+        )
+        self.log.info("Deleted Amazon ECR repository %s", self.repository_name)
+        return response
diff --git a/providers/amazon/src/airflow/providers/amazon/get_provider_info.py b/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
index a5a5c532e7e31..35fdd5756a6d1 100644
--- a/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
+++ b/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
@@ -94,6 +94,7 @@ def get_provider_info():
                 "integration-name": "Amazon Elastic Container Registry (ECR)",
                 "external-doc-url": "https://aws.amazon.com/ecr/",
                 "logo": "/docs/integration-logos/Amazon-Elastic-Container-Registry_light-bg@4x.png",
+                "how-to-guide": ["/docs/apache-airflow-providers-amazon/operators/ecr.rst"],
                 "tags": ["aws"],
             },
             {
@@ -424,6 +425,10 @@ def get_provider_info():
                 "integration-name": "Amazon EC2",
                 "python-modules": ["airflow.providers.amazon.aws.operators.ec2"],
             },
+            {
+                "integration-name": "Amazon Elastic Container Registry (ECR)",
+                "python-modules": ["airflow.providers.amazon.aws.operators.ecr"],
+            },
             {
                 "integration-name": "Amazon ECS",
                 "python-modules": ["airflow.providers.amazon.aws.operators.ecs"],
diff --git a/providers/amazon/tests/system/amazon/aws/example_ecr.py b/providers/amazon/tests/system/amazon/aws/example_ecr.py
new file mode 100644
index 0000000000000..64230657cb01e
--- /dev/null
+++ b/providers/amazon/tests/system/amazon/aws/example_ecr.py
@@ -0,0 +1,99 @@
+# 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.
+from __future__ import annotations
+
+from datetime import datetime
+
+from airflow.providers.amazon.aws.operators.ecr import (
+    EcrCreateRepositoryOperator,
+    EcrDeleteRepositoryOperator,
+    EcrSetRepositoryPolicyOperator,
+)
+from airflow.providers.common.compat.sdk import DAG, chain
+
+from system.amazon.aws.utils import ENV_ID_KEY, SystemTestContextBuilder
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS
+
+if AIRFLOW_V_3_0_PLUS:
+    from airflow.sdk import TriggerRule
+else:
+    from airflow.utils.trigger_rule import TriggerRule  # type: ignore[no-redef,attr-defined]
+
+DAG_ID = "example_ecr"
+
+sys_test_context_task = SystemTestContextBuilder().build()
+
+
+with DAG(
+    dag_id=DAG_ID,
+    schedule=None,
+    start_date=datetime(2024, 1, 1),
+    catchup=False,
+) as dag:
+    test_context = sys_test_context_task()
+    repository_name = f"{test_context[ENV_ID_KEY]}-test-repository"
+
+    # [START howto_operator_ecr_create_repository]
+    create_repository = EcrCreateRepositoryOperator(
+        task_id="create_repository",
+        repository_name=repository_name,
+    )
+    # [END howto_operator_ecr_create_repository]
+
+    # [START howto_operator_ecr_set_repository_policy]
+    set_repository_policy = EcrSetRepositoryPolicyOperator(
+        task_id="set_repository_policy",
+        repository_name=repository_name,
+        policy_text="""
+        {
+          "Version": "2012-10-17",
+          "Statement": [
+            {
+              "Sid": "AllowAccountPull",
+              "Effect": "Allow",
+              "Principal": {
+                "AWS": "arn:aws:iam::{{ task_instance.xcom_pull(task_ids='create_repository')['repository']['registryId'] }}:root"
+              },
+              "Action": [
+                "ecr:BatchGetImage",
+                "ecr:GetDownloadUrlForLayer"
+              ]
+            }
+          ]
+        }
+        """,
+    )
+    # [END howto_operator_ecr_set_repository_policy]
+
+    # [START howto_operator_ecr_delete_repository]
+    delete_repository = EcrDeleteRepositoryOperator(
+        task_id="delete_repository",
+        repository_name=repository_name,
+        force=True,
+        trigger_rule=TriggerRule.ALL_DONE,
+    )
+    # [END howto_operator_ecr_delete_repository]
+
+    chain(test_context, create_repository, set_repository_policy, delete_repository)
+
+    from tests_common.test_utils.watcher import watcher
+
+    list(dag.tasks) >> watcher()
+
+from tests_common.test_utils.system_tests import get_test_run  # noqa: E402
+
+test_run = get_test_run(dag)
diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_ecr.py b/providers/amazon/tests/unit/amazon/aws/operators/test_ecr.py
new file mode 100644
index 0000000000000..ba485ba4016d2
--- /dev/null
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_ecr.py
@@ -0,0 +1,208 @@
+# 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.
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+
+from airflow.providers.amazon.aws.hooks.ecr import EcrHook
+from airflow.providers.amazon.aws.operators.ecr import (
+    EcrCreateRepositoryOperator,
+    EcrDeleteRepositoryOperator,
+    EcrSetRepositoryPolicyOperator,
+)
+
+from unit.amazon.aws.utils.test_template_fields import validate_template_fields
+
+REPOSITORY_NAME = "test-repository"
+REGISTRY_ID = "123456789012"
+REPOSITORY_RESPONSE = {
+    "repository": {
+        "repositoryArn": f"arn:aws:ecr:us-east-1:{REGISTRY_ID}:repository/{REPOSITORY_NAME}",
+        "registryId": REGISTRY_ID,
+        "repositoryName": REPOSITORY_NAME,
+        "repositoryUri": f"{REGISTRY_ID}.dkr.ecr.us-east-1.amazonaws.com/{REPOSITORY_NAME}",
+    },
+    "ResponseMetadata": {"HTTPStatusCode": 200},
+}
+POLICY_TEXT = '{"Version":"2012-10-17","Statement":[]}'
+POLICY_RESPONSE = {
+    "registryId": REGISTRY_ID,
+    "repositoryName": REPOSITORY_NAME,
+    "policyText": POLICY_TEXT,
+    "ResponseMetadata": {"HTTPStatusCode": 200},
+}
+
+
+class TestEcrCreateRepositoryOperator:
+    @pytest.mark.parametrize(
+        ("operator_parameters", "boto3_parameters"),
+        [
+            pytest.param(
+                {},
+                {
+                    "repositoryName": REPOSITORY_NAME,
+                },
+                id="required-parameters",
+            ),
+            pytest.param(
+                {
+                    "registry_id": REGISTRY_ID,
+                    "tags": [{"Key": "environment", "Value": "test"}],
+                    "image_tag_mutability": "IMMUTABLE_WITH_EXCLUSION",
+                    "image_tag_mutability_exclusion_filters": [
+                        {"filterType": "WILDCARD", "filter": "latest"}
+                    ],
+                    "image_scanning_configuration": {"scanOnPush": True},
+                    "encryption_configuration": {
+                        "encryptionType": "KMS",
+                        "kmsKey": "arn:aws:kms:us-east-1:123456789012:key/test-key",
+                    },
+                },
+                {
+                    "repositoryName": REPOSITORY_NAME,
+                    "registryId": REGISTRY_ID,
+                    "tags": [{"Key": "environment", "Value": "test"}],
+                    "imageTagMutability": "IMMUTABLE_WITH_EXCLUSION",
+                    "imageTagMutabilityExclusionFilters": [{"filterType": "WILDCARD", "filter": "latest"}],
+                    "imageScanningConfiguration": {"scanOnPush": True},
+                    "encryptionConfiguration": {
+                        "encryptionType": "KMS",
+                        "kmsKey": "arn:aws:kms:us-east-1:123456789012:key/test-key",
+                    },
+                },
+                id="all-parameters",
+            ),
+        ],
+    )
+    @mock.patch.object(EcrHook, "conn", new_callable=mock.PropertyMock)
+    def test_execute(self, mock_conn, operator_parameters, boto3_parameters):
+        mock_client = mock.MagicMock(spec=["create_repository"])
+        mock_client.create_repository.return_value = REPOSITORY_RESPONSE
+        mock_conn.return_value = mock_client
+        operator = EcrCreateRepositoryOperator(
+            task_id="create_repository",
+            repository_name=REPOSITORY_NAME,
+            **operator_parameters,
+        )
+
+        result = operator.execute({})
+
+        mock_client.create_repository.assert_called_once_with(**boto3_parameters)
+        assert result == REPOSITORY_RESPONSE
+
+    def test_template_fields(self):
+        operator = EcrCreateRepositoryOperator(
+            task_id="create_repository",
+            repository_name=REPOSITORY_NAME,
+        )
+
+        validate_template_fields(operator)
+
+
+class TestEcrSetRepositoryPolicyOperator:
+    @pytest.mark.parametrize(
+        ("operator_parameters", "boto3_parameters"),
+        [
+            pytest.param(
+                {},
+                {
+                    "repositoryName": REPOSITORY_NAME,
+                    "policyText": POLICY_TEXT,
+                    "force": False,
+                },
+                id="defaults",
+            ),
+            pytest.param(
+                {"registry_id": REGISTRY_ID, "force": True},
+                {
+                    "registryId": REGISTRY_ID,
+                    "repositoryName": REPOSITORY_NAME,
+                    "policyText": POLICY_TEXT,
+                    "force": True,
+                },
+                id="registry-and-force",
+            ),
+        ],
+    )
+    @mock.patch.object(EcrHook, "conn", new_callable=mock.PropertyMock)
+    def test_execute(self, mock_conn, operator_parameters, boto3_parameters):
+        mock_client = mock.MagicMock(spec=["set_repository_policy"])
+        mock_client.set_repository_policy.return_value = POLICY_RESPONSE
+        mock_conn.return_value = mock_client
+        operator = EcrSetRepositoryPolicyOperator(
+            task_id="set_repository_policy",
+            repository_name=REPOSITORY_NAME,
+            policy_text=POLICY_TEXT,
+            **operator_parameters,
+        )
+
+        result = operator.execute({})
+
+        mock_client.set_repository_policy.assert_called_once_with(**boto3_parameters)
+        assert result == POLICY_RESPONSE
+
+    def test_template_fields(self):
+        operator = EcrSetRepositoryPolicyOperator(
+            task_id="set_repository_policy",
+            repository_name=REPOSITORY_NAME,
+            policy_text=POLICY_TEXT,
+        )
+
+        validate_template_fields(operator)
+
+
+class TestEcrDeleteRepositoryOperator:
+    @pytest.mark.parametrize(
+        ("operator_parameters", "boto3_parameters"),
+        [
+            pytest.param(
+                {},
+                {"repositoryName": REPOSITORY_NAME, "force": False},
+                id="defaults",
+            ),
+            pytest.param(
+                {"registry_id": REGISTRY_ID, "force": True},
+                {"repositoryName": REPOSITORY_NAME, "registryId": REGISTRY_ID, "force": True},
+                id="registry-and-force",
+            ),
+        ],
+    )
+    @mock.patch.object(EcrHook, "conn", new_callable=mock.PropertyMock)
+    def test_execute(self, mock_conn, operator_parameters, boto3_parameters):
+        mock_client = mock.MagicMock(spec=["delete_repository"])
+        mock_client.delete_repository.return_value = REPOSITORY_RESPONSE
+        mock_conn.return_value = mock_client
+        operator = EcrDeleteRepositoryOperator(
+            task_id="delete_repository",
+            repository_name=REPOSITORY_NAME,
+            **operator_parameters,
+        )
+
+        result = operator.execute({})
+
+        mock_client.delete_repository.assert_called_once_with(**boto3_parameters)
+        assert result == REPOSITORY_RESPONSE
+
+    def test_template_fields(self):
+        operator = EcrDeleteRepositoryOperator(
+            task_id="delete_repository",
+            repository_name=REPOSITORY_NAME,
+        )
+
+        validate_template_fields(operator)

From 9a02caaf354f8998cd21aa319d86a3858d2a74c4 Mon Sep 17 00:00:00 2001
From: AlejandroMorgante <62363051+AlejandroMorgante@users.noreply.github.com>
Date: Thu, 16 Jul 2026 18:01:04 -0300
Subject: [PATCH 2/2] Clarify ECR operator use cases

---
 providers/amazon/docs/operators/ecr.rst                        | 3 +++
 .../amazon/src/airflow/providers/amazon/aws/operators/ecr.py   | 2 --
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/providers/amazon/docs/operators/ecr.rst b/providers/amazon/docs/operators/ecr.rst
index 6c35efb869c28..466e3c72d5794 100644
--- a/providers/amazon/docs/operators/ecr.rst
+++ b/providers/amazon/docs/operators/ecr.rst
@@ -42,6 +42,7 @@ Create an ECR repository
 
 To create an ECR repository, use
 :class:`~airflow.providers.amazon.aws.operators.ecr.EcrCreateRepositoryOperator`.
+This operator can provision a repository before a workflow builds or pushes container images.
 The operator returns the complete response from the Boto3 ``create_repository`` API operation.
 
 .. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_ecr.py
@@ -57,6 +58,7 @@ Set an ECR repository policy
 
 To set the repository policy for an ECR repository, use
 :class:`~airflow.providers.amazon.aws.operators.ecr.EcrSetRepositoryPolicyOperator`.
+This operator can configure permissions such as cross-account access to images stored in the repository.
 The operator returns the complete response from the Boto3 ``set_repository_policy`` API operation.
 
 .. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_ecr.py
@@ -72,6 +74,7 @@ Delete an ECR repository
 
 To delete an ECR repository, use
 :class:`~airflow.providers.amazon.aws.operators.ecr.EcrDeleteRepositoryOperator`.
+This operator can clean up temporary repositories or repositories that are no longer needed.
 Set ``force=True`` to delete a repository that contains images. The operator returns the complete response
 from the Boto3 ``delete_repository`` API operation.
 
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/ecr.py b/providers/amazon/src/airflow/providers/amazon/aws/operators/ecr.py
index 738fafde86ad5..dec75ede9b582 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/operators/ecr.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/ecr.py
@@ -14,8 +14,6 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-"""Amazon Elastic Container Registry (ECR) operators."""
-
 from __future__ import annotations
 
 from collections.abc import Sequence