Github Stars

Call an external HTTP API from a task.

Calls a real external API — the GitHub REST API — from inside a task. The workflow loops over repositories and awaits get_stars for each, with the HTTP request wrapped in a task so its result is recorded. Reach for this pattern whenever a workflow needs to talk to a network service: keep the I/O in a task so a replay reuses the response instead of re-fetching it.

Run it

python examples/github_stars.py

from __future__ import annotations

import httpx

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow


@task
async def get_stars(repo: str):
    url = f"https://api.github.com/repos/{repo}"
    return httpx.get(url).json()["stargazers_count"]


@workflow
async def github_stars(ctx: ExecutionContext[list[str]]):
    if not ctx.input:
        raise TypeError("The list of repositories cannot be empty.")

    repos = ctx.input
    stars = {}
    for repo in repos:
        stars[repo] = await get_stars(repo)
    return stars


if __name__ == "__main__":  # pragma: no cover
    repositories = [
        "python/cpython",
        "microsoft/vscode",
        "localsend/localsend",
        "srush/GPU-Puzzles",
        "hyperknot/openfreemap",
    ]
    ctx = github_stars.run(repositories)
    print(ctx.to_json())

The httpx.get call lives inside get_stars, a @task. That placement is the point: a task’s return value is persisted, so on replay Flux serves the recorded star count rather than hitting the API again. The subflows example does the same job by promoting each fetch to its own subflow.