如何将任意文件SHA传入Airflow的GithubOperator？

高级任务 #

使用Airflow编排更新指定Github仓库中任意文件的内容。

已尝试的实现 #

Github API要求提供现有文件的SHA哈希值才能更新文件。已通过另一个GithubOperator实例获取该哈希值并存入XCom，当前核心问题是将该SHA传递给GithubOperator时，参数模板未被渲染，导致更新失败。也接受无需在任务间传递SHA的其他方案（比如用Python TaskFlow实现，但更希望用原生Operator解决）。

问题现象 #

运行代码时，日志显示模板未被渲染，随后因SHA类型断言失败（模板未解析导致不是字符串）报错：

[2023-06-01, 17:14:50 EDT] {{my_dag_file.py:33}} INFO - in 
main_result_processor the_sha is {{ task_instance.xcom_pull(task_ids='retrieve_xcom_blob_sha', dag_id='push_a_change_to_github', key='return_value') }}
[2023-06-01, 17:14:50 EDT] {{taskinstance.py:1768}} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/providers/github/operators/github.py", line 72, in execute
    return self.result_processor(github_result)
  File "/usr/local/airflow/dags/my_dag.py", line 59, in <lambda>
    result_processor=lambda repo: main_result_processor(repo, blob_sha),
  File "/usr/local/airflow/dags/my_dag.py", line 34, in main_result_processor
    repo.update_file(
  File "/usr/local/airflow/.local/lib/python3.10/site-packages/github/Repository.py", line 2220, in update_file
    assert isinstance(sha, str)
AssertionError

问题DAG代码 #

"""
Minimal case of what's failing -- since we can get as far as extracting the SHA, let's just
hardcode that SHA and see if we can get it into the GithubOperator.
"""

import json
import logging

from airflow.decorators import dag, task
from airflow.providers.github.operators.github import GithubOperator
from airflow.utils.dates import days_ago

PARTIAL_URL = "our-account/our-repo"
default_args = {}


@dag(
    default_args={},
    schedule_interval=None,
    start_date=days_ago(365),
    dag_id='push_a_change_to_github',
)
def push_a_change_to_github_main():
    dummy01_contents = json.dumps(
        {
            "revision": 1,
            "message": "This is a dummy JSON file",
        }
    )
    dummy_path = "some-dir/dummy01.json"
    def main_result_processor(repo, the_sha):
        logging.info(f"in main_result_processor the_sha is {the_sha}")
        repo.update_file(
            path=dummy_path,
            message=f"""Update {dummy_path}

            Automated commit by an Airflow DAG
            """,
            content=json.dumps(dummy01_contents),
            sha=the_sha,
        )

    @task(provide_context=True, retries=1)
    def retrieve_xcom_blob_sha():
        # We need this task in here because in the real world we can't just hardcode a
        # SHA at DAG scope. This forces us to use the XCom mechanism to return a value from a task.
        # Once it works we can replace this taskflow Python task with the GithubOperator task to
        # query Github for the SHA, which we've verified does work.
        some_sha = "abcd0123ef01abcdabcd0123ef01abcd456789ab"
        return some_sha

    blob_sha = retrieve_xcom_blob_sha()
    push_an_update = GithubOperator(
        task_id="push_an_update",
        retries=1,
        github_method="get_repo",
        github_method_args={"full_name_or_id": PARTIAL_URL, },
        result_processor=lambda repo: main_result_processor(repo, blob_sha),
    )


push_a_change_to_github_main()

问题核心在于：GithubOperator的result_processor参数不会被Airflow自动进行模板渲染，直接传递TaskFlow返回的XCom对象会导致模板字符串未被解析。以下两种方案可解决该问题：

方案1：用Python TaskFlow封装完整逻辑（推荐） #

将获取SHA和更新文件的逻辑放在同一个Python任务中，避免XCom传递的限制，代码更简洁易维护：

import json
import logging
from airflow.decorators import dag, task
from airflow.providers.github.hooks.github import GithubHook
from airflow.utils.dates import days_ago

PARTIAL_URL = "our-account/our-repo"
GITHUB_CONN_ID = "github_default"  # 替换为你的Github连接ID

@dag(
    schedule_interval=None,
    start_date=days_ago(365),
    dag_id='push_a_change_to_github',
)
def push_a_change_to_github_main():
    dummy_path = "some-dir/dummy01.json"
    dummy_contents = json.dumps(
        {
            "revision": 1,
            "message": "This is a dummy JSON file",
        }
    )

    @task(retries=1)
    def update_github_file():
        hook = GithubHook(github_conn_id=GITHUB_CONN_ID)
        github = hook.get_conn()
        repo = github.get_repo(PARTIAL_URL)
        
        # 获取目标文件的SHA（真实场景中可根据需求调整逻辑）
        file_content = repo.get_contents(dummy_path)
        file_sha = file_content.sha
        
        # 执行文件更新
        repo.update_file(
            path=dummy_path,
            message=f"Update {dummy_path}\n\nAutomated commit by Airflow DAG",
            content=dummy_contents,
            sha=file_sha,
        )

    update_github_file()

push_a_change_to_github_main()

方案2：在result_processor中手动拉取XCom #

如果坚持使用GithubOperator，可在result_processor中直接从当前TaskInstance拉取XCom值：

import json
import logging
from airflow.decorators import dag, task
from airflow.providers.github.operators.github import GithubOperator
from airflow.utils.dates import days_ago
from airflow.models import TaskInstance

PARTIAL_URL = "our-account/our-repo"
default_args = {}


@dag(
    default_args={},
    schedule_interval=None,
    start_date=days_ago(365),
    dag_id='push_a_change_to_github',
)
def push_a_change_to_github_main():
    dummy01_contents = json.dumps(
        {
            "revision": 1,
            "message": "This is a dummy JSON file",
        }
    )
    dummy_path = "some-dir/dummy01.json"
    
    def main_result_processor(repo, task_instance):
        # 手动拉取XCom中的SHA值
        the_sha = task_instance.xcom_pull(task_ids='retrieve_xcom_blob_sha', key='return_value')
        logging.info(f"in main_result_processor the_sha is {the_sha}")
        repo.update_file(
            path=dummy_path,
            message=f"""Update {dummy_path}

            Automated commit by an Airflow DAG
            """,
            content=dummy01_contents,  # 注意：原代码中已做json.dumps，无需重复序列化
            sha=the_sha,
        )

    @task(retries=1)
    def retrieve_xcom_blob_sha():
        some_sha = "abcd0123ef01abcdabcd0123ef01abcd456789ab"
        return some_sha

    blob_sha = retrieve_xcom_blob_sha()
    push_an_update = GithubOperator(
        task_id="push_an_update",
        retries=1,
        github_method="get_repo",
        github_method_args={"full_name_or_id": PARTIAL_URL, },
        # 传递当前TaskInstance到result_processor
        result_processor=lambda repo, ti=TaskInstance.current(): main_result_processor(repo, ti),
    )


push_a_change_to_github_main()

内容的提问来源于stack exchange，提问作者Philip