操作器¶

使用 GithubOperator 在 GitHub 中执行操作。

您可以使用 GithubOperator 并从顶级 PyGithub 方法传递 github_method 和 github_method_args 来构建您自己的操作器。

您可以使用 result_processor 可调用对象进一步处理结果，就像您喜欢的那样。

列出用户拥有的所有存储库的示例，client.get_user().get_repos() 可以按如下方式实现

tests/system/providers/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/providers/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/providers/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/providers/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,
)