prefect_flow_task_github_repo_stats_fetcher.py

1import httpx
2from prefect import flow, task
3
4
5@task(log_prints=True)
6def get_repo_info(repo_name: str):
7    url = f"https://api.github.com/repos/{repo_name}"
8    response = httpx.get(url)
9    response.raise_for_status()
10    repo = response.json()
11    print(f"{repo_name} repository statistics 统计:")
12    print(f"Stars 🌠 : {repo['stargazers_count']}")
13    print(f"Forks 🍴 : {repo['forks_count']}")
14
15
16@flow
17def get_repo_stats(repos: list[str]):
18    for repo in repos:
19        get_repo_info(repo)
20
21
22if __name__ == "__main__":
23    get_repo_stats(["PrefectHQ/prefect", "pydantic/pydantic"])