Collecting one API is easy. The engineering question begins when three sources disagree about pagination, JSON structure, and metrics, yet must enter one reliable flow—and a second run must update rather than blindly replace the first.
This project collects OpenAlex works, Crossref metadata, and GitHub repositories. By default it requests two pages of five records from each source and normalizes them into seven columns: source, external_id, title, url, record_type, metric_name, and metric_value.
Make pagination differences explicit jobs
OpenAlex and GitHub use pages while Crossref uses offsets. jobs.hhy represents that difference instead of hiding it:
for page in range(1, config.pages + 1) {
jobs = append(jobs, { source: "OpenAlex", page: page, offset: 0 })
jobs = append(jobs, { source: "Crossref", page: page, offset: (page - 1) * config.per_page })
jobs = append(jobs, { source: "GitHub", page: page, offset: 0 })
}
Six jobs enter parallel(3). Each waits 250 ms before its request and uses a ten-second timeout, two retries, and 500 ms backoff. This is not a precise global rate limiter, but it prevents the example from starting with a burst against public APIs.
Unauthenticated GitHub search has a low quota. A production flow should reduce frequency or inject authenticated requests through a safe process extension, not store a token in project configuration.
Normalize shape without pretending semantics match
Each API has separate request and normalization functions. OpenAlex reports citations, Crossref references, and GitHub stars. They share the metric_name/metric_value shape without being collapsed into an ambiguous score.
OpenAlex → work id / display_name / citations
Crossref → DOI / title / references
GitHub → full_name / repository / stars
↓
source / external_id / title / url / type / metric_name / metric_value
A common schema simplifies storage, deduplication, and consumption. Retaining the metric name preserves source meaning.
Parsing can fail after a successful download, so network and normalization stages have separate attempt boundaries. Failures retain source, page, and error in failures.json. Successful pages can still merge, while the overall report becomes unsuccessful and the process returns 1.
Incremental merge requires stable identity
The collector loads existing CSV when present and flattens old plus new records into one Stream. Identity is source + external_id:
[existing, incoming]
|> stream
|> flat_map { records -> records |> stream }
|> group_by { record -> join([record.source, record.external_id], ":") }
|> map { group -> group.values[length(group.values) - 1] }
|> sort_by { record -> join([record.source, record.external_id], ":") }
|> collect
Old records come first and incoming records second, so the last value naturally replaces a matching key. Sorting by the same identity prevents API ordering changes from producing meaningless diffs.
CSV, the run report, and the failure list are all written atomically. Every record retains its source URL for verification.
Why the test runs twice
The local fixture first returns twelve cross-page inputs that deduplicate to nine unique records. The second run loads those nine, collects again, replaces matching identities, and still ends with nine.
sh practical-projects/multi-api-data-collector/self-test.sh
One run proves that CSV export works. Two runs verify composite identity, replacement direction, stable ordering, and an idempotent shape. The fixture uses a random local port and temporary directory, then removes itself without contaminating production output.
The boundary of this flow
parallel improves HTTP waiting; it is not a claim about numeric throughput. retry handles transient failure; it cannot repair an incompatible schema. Incremental replacement preserves the latest observation; it is not full history.
With those boundaries explicit, the project shows HHY's intended direction: a small set of composable network, Stream, error, and atomic-file primitives that keep reliability policy inside the business flow.