操作符¶

使用 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,
)