運算子¶

使用 GithubOperator 在 GitHub 中執行操作。

您可以使用 GithubOperator 並透過傳遞頂級 PyGithub 方法中的 github_method 和 github_method_args 來構建自己的運算子。

您可以進一步使用 result_processor 可呼叫物件根據需要處理結果。

列出使用者擁有的所有倉庫的示例，client.get_user().get_repos() 可以按如下方式實現

tests/system/github/example_github.py

github_list_repos = GithubOperator(
    task_id="github_list_repos",
    github_method="get_user",
    result_processor=lambda user: logger.info(list(user.get_repos())),
)

列出倉庫中標籤的示例，client.get_repo(full_name_or_id=’apache/airflow’).get_tags() 可以按如下方式實現

tests/system/github/example_github.py

list_repo_tags = GithubOperator(
    task_id="list_repo_tags",
    github_method="get_repo",
    github_method_args={"full_name_or_id": "apache/airflow"},
    result_processor=lambda repo: logger.info(list(repo.get_tags())),
)

感測器¶

您可以使用 GithubSensor 構建自己的感測器，

您還可以使用 BaseGithubRepositorySensor 在倉庫上實現自己的感測器，一個示例是 GithubTagSensor

使用 GithubTagSensor 等待在 GitHub 中建立標籤。

標籤 v1.0 的示例

tests/system/github/example_github.py

tag_sensor = GithubTagSensor(
    task_id="example_tag_sensor",
    tag_name="v1.0",
    repository_name="apache/airflow",
    timeout=60,
    poke_interval=10,
)

透過直接使用 GithubSensor 可以實現類似的功能。

tests/system/github/example_github.py

def tag_checker(repo: Any, tag_name: str) -> bool | None:
    result = None
    try:
        if repo is not None and tag_name is not None:
            all_tags = [x.name for x in repo.get_tags()]
            result = tag_name in all_tags

    except GithubException as github_error:  # type: ignore[misc]
        raise AirflowException(f"Failed to execute GithubSensor, error: {github_error}")
    except Exception as e:
        raise AirflowException(f"GitHub operator error: {e}")
    return result

github_sensor = GithubSensor(
    task_id="example_sensor",
    method_name="get_repo",
    method_params={"full_name_or_id": "apache/airflow"},
    result_processor=lambda repo: tag_checker(repo, "v1.0"),
    timeout=60,
    poke_interval=10,
)