synapse-data-curation

Synapse Data Curation

Overview

Advanced patterns for curating data in Synapse (synapse.org) via the REST API. Covers the annotations2 endpoint with etag/optimistic-locking, Dataset entity creation with dynamically-derived column schemas, JSON Schema binding for metadata validation, team permissions, and wiki management. Complements the synapse-python-client skill, which covers the high-level synapseclient.models SDK.

Installation

pip install synapseclient pyyaml

Authentication

import synapseclient

# Via environment variable (recommended for automated workflows)
syn = synapseclient.Synapse()
syn.login(authToken=os.environ["SYNAPSE_AUTH_TOKEN"], silent=True)

# Via cached credentials (~/.synapseConfig)
syn = synapseclient.Synapse()
syn.login(silent=True)

Core Concepts

• annotations2: The REST endpoint for reading and writing entity annotations. Uses etag-based optimistic locking — always re-fetch before PUT.
• Dataset entity: org.sagebionetworks.repo.model.table.Dataset — a typed table entity that must be a direct child of its project and whose columnIds should be derived dynamically from the annotation keys present on its files.
• Schema binding: JSON Schema is bound to a files folder (not the Dataset entity) to enable validation in Curator Grid.
• 412 Precondition Failed: The server rejects a PUT when your etag is stale. Always retry after re-fetching.

Common Operations

Reading and Writing Annotations

import json, time

def set_annotations(syn, entity_id: str, fields: dict, max_attempts: int = 3):
    """Set annotations on an entity, handling 412 stale-etag retries."""
    for attempt in range(max_attempts):
        try:
            ann = syn.restGET(f"/entity/{entity_id}/annotations2")
            for k, v in fields.items():
                if isinstance(v, list):
                    ann["annotations"][k] = {"type": "STRING", "value": v}
                elif isinstance(v, str):
                    ann["annotations"][k] = {"type": "STRING", "value": [v]}
                elif isinstance(v, int):
                    ann["annotations"][k] = {"type": "INTEGER", "value": [v]}
                elif isinstance(v, float):
                    ann["annotations"][k] = {"type": "DOUBLE", "value": [v]}
            syn.restPUT(f"/entity/{entity_id}/annotations2", json.dumps(ann))
            return
        except Exception as e:
            if "412" in str(e) and attempt < max_attempts - 1:
                time.sleep(1)
            else:
                raise

# Remove a single field
ann = syn.restGET(f"/entity/{entity_id}/annotations2")
ann["annotations"].pop("fieldName", None)
syn.restPUT(f"/entity/{entity_id}/annotations2", json.dumps(ann))