InvenioRDM

Read records and files from InvenioRDM repositories, such as Zenodo and INIS

Zenodo and INIS both run InvenioRDM and share its REST API. Each version of a dataset is a separate record with its own id. The files of a published record never change, so a record id and a file name identify the exact content of a file.

from fastcore.test import *

---

source

fetch_file

def fetch_file(
    record_id:int, # Record id
    key:str, # File name, as the record's `files` list gives it
    base_url:str='https://zenodo.org/api', # API root of the repository
)->bytes: # File content

Download file key of record record_id

fetch_file downloads one file of a Zenodo record by default. Record 19387002 holds the Fram Strait 2020-2021 measurements as a CSV file:

content = fetch_file(19387002, 'FramStrait_2020_2021_radionuclides.csv')
test_eq(content[:15], b'Cruise,Station,')

A file name can include a folder. The JOIS 2021 record names its only file annabel-payne/BGOS-JOIS-2021-v1.0.1.zip:

content = fetch_file(18880401, 'annabel-payne/BGOS-JOIS-2021-v1.0.1.zip')
test_eq(content[:2], b'PK')

Records and versions

---

source

fetch_record

def fetch_record(
    record_id, # Record id
    base_url:str='https://zenodo.org/api', # API root of the repository
)->dict | None: # Record as InvenioRDM JSON, or None if it does not exist

Fetch record record_id

fetch_record returns a record in the InvenioRDM JSON format, which Zenodo and INIS share. parent.id identifies the dataset across all its versions. Zenodo calls it the conceptrecid. versions tells whether the record is the latest version. Record 20761705 is the first version of the Fram Strait 2024 I-129 dataset:

rec = fetch_record(20761705)
test_eq(rec['parent']['id'], '20761704')
test_eq(rec['versions'], {'is_latest': False, 'index': 1})

A record that does not exist gives None:

test_eq(fetch_record(999999999), None)

---

source

fetch_latest

def fetch_latest(
    record_id, # Id of any version of the dataset
    base_url:str='https://zenodo.org/api', # API root of the repository
)->dict | None: # Latest version as InvenioRDM JSON, or None if the record does not exist

Fetch the latest version of the dataset that record record_id belongs to

fetch_latest finds out whether a dataset has a newer version than the record a handler reads. The Fram Strait 2024 dataset has a second version, record 23111574, which adds U-236. Record ids in InvenioRDM JSON are strings:

test_eq(fetch_latest(20761705)['id'], '23111574')

Communities

---

source

community_records

def community_records(
    slug:str, # Community name in its URL, such as 'titanica'
    base_url:str='https://zenodo.org/api', # API root of the repository
)->list[dict]: # Records as InvenioRDM JSON, newest first

Every record in community slug

community_records requests pages of 25 records, the most Zenodo returns without a token, and follows each page’s next link. A community lists only the latest version of each dataset, so every record has a different parent.id:

recs = community_records('titanica')
test_eq(len({r['parent']['id'] for r in recs}), len(recs))
len(recs)

A misspelt community raises an error, rather than returning no records:

with expect_fail(ValueError, contains='titanic'): community_records('titanic')