Pythonスクリプトをスケジューリングやリトライ対応のデータパイプラインに変えるワークフレームワーク

pip install -U prefect

uv add prefect

from prefect import flow, task
import httpx


@task(log_prints=True)
def get_stars(repo: str):
    url = f"https://api.github.com/repos/{repo}"
    count = httpx.get(url).json()["stargazers_count"]
    print(f"{repo} has {count} stars!")


@flow(name="GitHub Stars")
def github_stars(repos: list[str]):
    for repo in repos:
        get_stars(repo)


if __name__ == "__main__":
    github_stars(["PrefectHQ/prefect"])

prefect server start

if __name__ == "__main__":
    github_stars.serve(
        name="first-deployment",
        cron="* * * * *",
        parameters={"repos": ["PrefectHQ/prefect"]}
    )