prefect_flow_task_github_repo_stats_quickstart.py

1import httpx
2from prefect import flow, task
3
4
5@task(log_prints=True)
6def get_repo_info(repo_owner: str, repo_name: str):
7    """Get info about a repo - will retry twice after failing"""
8    url = f"https://api.github.com/repos/{repo_owner}/{repo_name}"
9    api_response = httpx.get(url)
10    api_response.raise_for_status()
11    repo_info = api_response.json()
12    print(f"Stars: {repo_info['stargazers_count']}")
13    print(f"Forks: {repo_info['forks_count']}")
14
15
16@flow
17def get_stats(repo_owner: str = "PrefectHQ", repo_name: str = "prefect"):
18    """
19    The core logic of the workflow.
20    Tasks are called within the flow.
21    """
22    get_repo_info(repo_owner, repo_name)
23
24
25if __name__ == "__main__":
26    get_stats()