使用 Apache Airflow Amazon 提供方管理 Amazon ECR 仓库创建、策略与删除 Operator 实战指南【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowAmazon Elastic Container RegistryECR是 AWS 提供的托管容器镜像仓库服务用于存储、共享和部署容器镜像与制品。本文基于 Apache Airflow 官方 amazon 提供方apache-airflow-providers-amazon的 ECR Operator 文档系统讲解EcrCreateRepositoryOperator、EcrSetRepositoryPolicyOperator、EcrDeleteRepositoryOperator三个 Operator 的完整用法并深入到本仓库源码与测试揭示其底层 Boto3 调用链、模板化字段与返回值结构。读完本文你将能够在 Airflow DAG 中完整编排创建仓库 → 设置跨账号拉取策略 → 清理删除仓库的容器镜像发布生命周期。前置任务Prerequisite Tasks在 DAG 中使用 ECR Operator 之前需要完成三件事创建必要的 AWS 资源通过 AWS Console 或 AWS CLI 预先准备好账号、IAM 权限等基础设施安装 API 库通过 pip 安装带amazonextra 的 Airflow 包pip install apache-airflow[amazon]安装后即可导入本仓库中提供的 ECR 相关类详细安装指引见 Airflow 安装文档当前仓库中对应源码位于 providers/amazon/src/airflow/providers/amazon/aws/operators/ecr.py。配置 AWS Connection为 Airflow 配置 AWS 连接使 Operator 能获取访问 ECR API 所需的凭证具体配置方法见 AWS Connection 配置文档。通用参数Generic Parameters所有 ECR Operator 均继承自AwsBaseOperator因此共享一套 AWS 通用参数原始说明见 generic_parameters.rst参数说明默认值aws_conn_id引用 AWS Connection 的 ID。若设为None则走 Boto3 默认行为不查找连接直接使用环境中配置的凭证否则使用 Connection 中保存的凭证。注意在分布式部署下若为NoneBoto3 默认配置需在每个 worker 节点上维护aws_defaultregion_nameAWS 区域名。若为None或省略则使用 AWS Connection Extra 参数中的region_name否则覆盖连接值Noneverify是否校验 SSL 证书。False表示不校验也可以传 CA 证书 bundle 文件路径用于替换 botocore 默认证书。为None时沿用连接 Extra 参数中的verify值Nonebotocore_config用于构造botocore.config.Config的字典可配置重试策略、超时等以避免限流异常。为None时使用连接 Extra 参数中的config_kwargs。注意传空字典{}会覆盖连接级配置Nonebotocore_config支持的核心配置示例botocore_config{ signature_version: unsigned, s3: { us_east_1_regional_endpoint: True, }, retries: { mode: standard, max_attempts: 10, }, connect_timeout: 300, read_timeout: 300, tcp_keepalive: True, }其中retries.mode可设为standard标准重试模式或adaptivemax_attempts控制总重试次数connect_timeout与read_timeout单位为秒。创建 ECR 仓库EcrCreateRepositoryOperator基本用法在构建或推送容器镜像之前工作流可以先调用EcrCreateRepositoryOperator预置目标仓库。该 Operator 会调用 Boto3 的create_repositoryAPI并返回该 API 的完整响应包含repository对象其中有repositoryArn、registryId、repositoryName、repositoryUri等关键字段。from airflow.providers.amazon.aws.operators.ecr import EcrCreateRepositoryOperator create_repository EcrCreateRepositoryOperator( task_idcreate_repository, repository_namemy-app-repository, )完整参数与模板化字段从源码 operators/ecr.py 可以看到该 Operator 的构造参数与 Boto3create_repository请求参数一一对应参数类型说明repository_namestr必填要创建的仓库名称支持 Jinja 模板渲染registry_idstr \| None仓库所属的 AWS 账户 ID默认使用当前凭证账户tagslist[dict[str, str]] \| None应用到仓库的元数据标签如[{Key: environment, Value: test}]image_tag_mutabilitystr \| None镜像标签可变性设置可选MUTABLE、IMMUTABLE、IMMUTABLE_WITH_EXCLUSIONimage_tag_mutability_exclusion_filterslist[dict[str, str]] \| None覆盖仓库标签可变性设置的过滤器如[{filterType: WILDCARD, filter: latest}]image_scanning_configurationdict[str, bool] \| None镜像扫描配置如{scanOnPush: True}推送时自动扫描漏洞encryption_configurationdict[str, str] \| None加密配置如{encryptionType: KMS, kmsKey: arn:aws:kms:...}encryptionType可选AES256或KMS上述除aws_conn_id外的业务参数均声明在template_fields中意味着可在 DAG 中使用 Jinja 模板动态生成其中tags、image_tag_mutability_exclusion_filters、image_scanning_configuration、encryption_configuration的渲染器为json模板渲染后按 JSON 解析。底层实现解析execute()方法的核心逻辑非常简洁见 operators/ecr.pydef 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要点self.hook是EcrHook定义于 hooks/ecr.pyself.hook.conn返回一个已按连接凭证初始化的 Boto3 ECR 客户端prune_dict会剔除值为None的参数因此未显式指定的可选参数不会被发送给 AWS避免 API 校验失败方法返回值为 Boto3 响应字典可通过 XCom 供下游任务使用。设置 ECR 仓库策略EcrSetRepositoryPolicyOperator基本用法EcrSetRepositoryPolicyOperator用于设置 ECR 仓库的访问策略典型场景是配置跨账号访问——允许其他 AWS 账号拉取本仓库中的镜像。它调用 Boto3 的set_repository_policyAPI并返回完整响应包含registryId、repositoryName、policyText字段。from airflow.providers.amazon.aws.operators.ecr import EcrSetRepositoryPolicyOperator set_repository_policy EcrSetRepositoryPolicyOperator( task_idset_repository_policy, repository_namemy-app-repository, policy_text { Version: 2012-10-17, Statement: [ { Sid: AllowAccountPull, Effect: Allow, Principal: { AWS: arn:aws:iam::123456789012:root }, Action: [ ecr:BatchGetImage, ecr:GetDownloadUrlForLayer ] } ] } , )参数说明从源码 operators/ecr.py 可知其构造参数参数类型说明repository_namestr必填要设置策略的仓库名称支持模板渲染policy_textstr必填要应用的 JSON 格式仓库策略文本支持模板渲染渲染器为jsonregistry_idstr \| None仓库所属的 AWS 账户 IDforcebool是否在现有策略阻止设置新策略时强制覆盖默认False与 XCom 联动实现动态跨账号授权本仓库的系统测试 example_ecr.py 展示了一个非常实用的技巧利用 Jinja 模板 XCom把创建仓库任务返回的registryId动态注入到策略的Principal中从而实现谁创建了仓库谁就有权拉取的自动化授权policy_text { Version: 2012-10-17, Statement: [ { Sid: AllowAccountPull, Effect: Allow, Principal: { AWS: arn:aws:iam::{{ task_instance.xcom_pull(task_idscreate_repository)[repository][registryId] }}:root }, Action: [ ecr:BatchGetImage, ecr:GetDownloadUrlForLayer ] } ] } 由于policy_text位于template_fields{{ task_instance.xcom_pull(...) }}会在运行时被渲染为上一个任务返回的实际registryId无需硬编码账号 ID。删除 ECR 仓库EcrDeleteRepositoryOperator基本用法EcrDeleteRepositoryOperator用于清理临时仓库或不再需要的仓库。它调用 Boto3 的delete_repositoryAPI并返回完整响应。from airflow.providers.amazon.aws.operators.ecr import EcrDeleteRepositoryOperator from airflow.sdk import TriggerRule # Airflow 3.x2.x 使用 from airflow.utils.trigger_rule import TriggerRule delete_repository EcrDeleteRepositoryOperator( task_iddelete_repository, repository_namemy-app-repository, forceTrue, trigger_ruleTriggerRule.ALL_DONE, )参数说明从源码 operators/ecr.py 可知其构造参数参数类型说明repository_namestr必填要删除的仓库名称支持模板渲染registry_idstr \| None仓库所属的 AWS 账户 IDforcebool仓库内仍有镜像时是否强制删除默认False。注意ECR 不允许直接删除非空仓库若仓库包含镜像且forceFalsedelete_repository会抛出RepositoryNotEmptyException一个值得注意的细节在官方系统测试中删除任务被设置为forceTrue且trigger_ruleTriggerRule.ALL_DONE。ALL_DONE意味着无论上游设置策略任务成功还是失败删除任务都会执行。这是一个良好的清理型任务实践——确保在测试或 CI 场景中仓库总能被清理避免残留资源产生费用。完整示例 DAGECR 仓库生命周期编排综合以上三个 Operator即可编排出一条完整的创建 → 授权 → 清理流水线。以下示例 DAG 来自本仓库官方系统测试 example_ecr.py它覆盖了本文介绍的全部三个 Operatorfrom datetime import datetime from airflow.providers.amazon.aws.operators.ecr import ( EcrCreateRepositoryOperator, EcrDeleteRepositoryOperator, EcrSetRepositoryPolicyOperator, ) from airflow.providers.common.compat.sdk import DAG, chain from airflow.sdk import TriggerRule DAG_ID example_ecr with DAG( dag_idDAG_ID, scheduleNone, start_datedatetime(2024, 1, 1), catchupFalse, ) as dag: repository_name test-repository create_repository EcrCreateRepositoryOperator( task_idcreate_repository, repository_namerepository_name, ) set_repository_policy EcrSetRepositoryPolicyOperator( task_idset_repository_policy, repository_namerepository_name, policy_text { Version: 2012-10-17, Statement: [ { Sid: AllowAccountPull, Effect: Allow, Principal: { AWS: arn:aws:iam::{{ task_instance.xcom_pull(task_idscreate_repository)[repository][registryId] }}:root }, Action: [ ecr:BatchGetImage, ecr:GetDownloadUrlForLayer ] } ] } , ) delete_repository EcrDeleteRepositoryOperator( task_iddelete_repository, repository_namerepository_name, forceTrue, trigger_ruleTriggerRule.ALL_DONE, ) chain(create_repository, set_repository_policy, delete_repository)在真实生产场景中你通常会把镜像构建如DockerOperator或 ECS/Batch 构建任务插入set_repository_policy与delete_repository之间将删除任务降级为失败时的兜底清理。源码级验证单元测试如何保证行为仓库中的单元测试 test_ecr.py 通过参数化用例验证了 Operator 与 Boto3 API 的映射关系最小参数仅传repository_name时create_repository只接收{repositoryName: ...}一个参数证明prune_dict确实剔除了所有None参数全量参数传入registry_id、tags、image_tag_mutability、image_tag_mutability_exclusion_filters、image_scanning_configuration、encryption_configuration后断言 Boto3 客户端被以完全一致的驼峰命名参数调用返回值断言execute()的返回结果等于 Boto3 客户端的原始响应字典验证了返回 Boto3 完整响应的文档承诺模板字段调用validate_template_fields(operator)校验template_fields声明完整。测试中还给出了 ECR 响应结构的真实形态便于理解 XCom 取值的路径REPOSITORY_RESPONSE { repository: { repositoryArn: arn:aws:ecr:us-east-1:123456789012:repository/test-repository, registryId: 123456789012, repositoryName: test-repository, repositoryUri: 123456789012.dkr.ecr.us-east-1.amazonaws.com/test-repository, }, ResponseMetadata: {HTTPStatusCode: 200}, }这解释了为何示例 DAG 中要用xcom_pull(task_idscreate_repository)[repository][registryId]取值——registryId嵌套在响应字典的repository键下。延伸阅读ECR Operator 完整源码providers/amazon/src/airflow/providers/amazon/aws/operators/ecr.pyEcrHook 与临时凭证处理EcrCredentials支持docker login格式的 registry 地址providers/amazon/src/airflow/providers/amazon/aws/hooks/ecr.py官方系统测试示例 DAGproviders/amazon/tests/system/amazon/aws/example_ecr.py单元测试providers/amazon/tests/unit/amazon/aws/operators/test_ecr.pyAWS Connection 配置providers/amazon/docs/connections/aws.rst原始文档本文主题来源providers/amazon/docs/operators/ecr.rst【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考