Automating Zenodo Batch Uploads with the REST API #

from __future__ import annotations

import asyncio
import hashlib
from pathlib import Path
from typing import Any

import httpx

ZENODO_API = "https://sandbox.zenodo.org/api"  # swap for https://zenodo.org/api in production


def compute_md5(path: Path, chunk: int = 1 << 20) -> str:
    """Stream the file through MD5 so large uploads stay memory-constant."""
    digest = hashlib.md5()
    with path.open("rb") as fh:
        for block in iter(lambda: fh.read(chunk), b""):
            digest.update(block)
    return digest.hexdigest()


class ZenodoError(RuntimeError):
    """Raised on an unrecoverable Zenodo API response."""


async def create_deposition(client: httpx.AsyncClient) -> dict[str, Any]:
    """POST an empty deposition; the response carries links.bucket and the id."""
    resp = await client.post("/api/deposit/depositions", json={})
    resp.raise_for_status()
    return resp.json()


async def _put_file(
    client: httpx.AsyncClient, bucket_url: str, path: Path, sem: asyncio.Semaphore
) -> None:
    """Stream one file to the bucket and verify Zenodo's returned checksum."""
    local_md5 = compute_md5(path)
    async with sem:
        with path.open("rb") as body:
            resp = await client.put(f"{bucket_url}/{path.name}", content=body)
    resp.raise_for_status()
    remote = resp.json().get("checksum", "").removeprefix("md5:")
    if remote.lower() != local_md5.lower():
        raise ZenodoError(f"Checksum mismatch for {path.name}: {remote} != {local_md5}")


async def upload_files(
    client: httpx.AsyncClient, bucket_url: str, files: list[Path], concurrency: int = 4
) -> None:
    """Upload a batch of files with bounded concurrency."""
    sem = asyncio.Semaphore(concurrency)
    await asyncio.gather(*(_put_file(client, bucket_url, f, sem) for f in files))


async def set_metadata(
    client: httpx.AsyncClient, deposition_id: int, metadata: dict[str, Any]
) -> None:
    """PUT the deposition metadata; a 400 lists the offending fields."""
    resp = await client.put(
        f"/api/deposit/depositions/{deposition_id}", json={"metadata": metadata}
    )
    if resp.status_code == 400:
        raise ZenodoError(f"Metadata rejected: {resp.json()}")
    resp.raise_for_status()


async def publish(client: httpx.AsyncClient, deposition_id: int) -> dict[str, Any]:
    """POST the publish action; the response carries the minted DOI."""
    resp = await client.post(f"/api/deposit/depositions/{deposition_id}/actions/publish")
    resp.raise_for_status()
    return resp.json()


async def deposit_batch(
    token: str, files: list[Path], metadata: dict[str, Any]
) -> dict[str, str]:
    """Full flow: create -> upload -> set metadata -> publish -> capture DOI."""
    headers = {"Authorization": f"Bearer {token}"}
    async with httpx.AsyncClient(base_url=ZENODO_API, headers=headers, timeout=120.0) as client:
        deposition = await create_deposition(client)
        dep_id = deposition["id"]
        bucket = deposition["links"]["bucket"]
        await upload_files(client, bucket, files)
        await set_metadata(client, dep_id, metadata)
        published = await publish(client, dep_id)
        return {"doi": published["doi"], "concept_doi": published.get("conceptdoi", "")}


async def new_version(
    client: httpx.AsyncClient, published_deposition_id: int
) -> dict[str, Any]:
    """Start a new version from a published record and return the fresh draft."""
    resp = await client.post(
        f"/api/deposit/depositions/{published_deposition_id}/actions/newversion"
    )
    resp.raise_for_status()
    draft_url = resp.json()["links"]["latest_draft"]
    draft = await client.get(draft_url)
    draft.raise_for_status()
    return draft.json()  # replace changed files on its bucket, then publish again


if __name__ == "__main__":
    meta = {
        "upload_type": "dataset",
        "title": "Reef Temperature Series 2019–2024",
        "creators": [{"name": "Curie, Marie", "orcid": "0000-0002-1825-0097"}],
        "description": "<p>Hourly sea-surface temperatures from 12 reef stations.</p>",
        "access_right": "open",
        "license": "cc-by-4.0",
        "keywords": ["oceanography", "sea-surface temperature"],
    }
    result = asyncio.run(
        deposit_batch("YOUR_TOKEN", [Path("reef_sst.csv"), Path("stations.json")], meta)
    )
    print(result["doi"])

import asyncio
from pathlib import Path

import httpx
import pytest


def _handler(request: httpx.Request) -> httpx.Response:
    if request.method == "PUT" and "/files/" in request.url.path:
        # Simulate Zenodo returning a wrong checksum.
        return httpx.Response(200, json={"checksum": "md5:" + "0" * 32})
    return httpx.Response(200, json={})


@pytest.mark.asyncio
async def test_bad_checksum_is_rejected(tmp_path: Path) -> None:
    f = tmp_path / "data.csv"
    f.write_bytes(b"a,b\n1,2\n")
    transport = httpx.MockTransport(_handler)
    async with httpx.AsyncClient(base_url="https://sandbox.zenodo.org/api",
                                 transport=transport) as client:
        with pytest.raises(ZenodoError):
            await upload_files(client, "https://sandbox.zenodo.org/api/files/abc", [f])