Rest APIs

This section is for super users / developers only. These functions are subject to change as they are internal development functions. Use at your own risk.

The functions provided in these documents are generally direct representations of the Synapse REST API. They're used within the Synapse Python Client to interact with the Synapse servers. These will eventually be removed from the Synapse Python Client in favor of an auto-generated client derived from the Synapse Open API Spec.

synapseclient.api.agent_services

This module is responsible for exposing the services defined at: https://rest-docs.synapse.org/rest/#org.sagebionetworks.repo.web.controller.AgentController

Classes

Functions

register_agent async

register_agent(cloud_agent_id: str, cloud_alias_id: Optional[str] = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Registers an agent with Synapse OR gets existing agent registration. Sends a request matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/AgentRegistrationRequest.html

PARAMETER	DESCRIPTION
cloud_agent_id	The cloud provider ID of the agent to register. TYPE: str
cloud_alias_id	The cloud provider alias ID of the agent to register. In the Synapse API, this defaults to 'TSTALIASID'. TYPE: Optional[str] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The registered agent matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/AgentRegistration.html

Source code in synapseclient/api/agent_services.py

async def register_agent(
    cloud_agent_id: str,
    cloud_alias_id: Optional[str] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Registers an agent with Synapse OR gets existing agent registration.
    Sends a request matching
    <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/AgentRegistrationRequest.html>

    Arguments:
        cloud_agent_id: The cloud provider ID of the agent to register.
        cloud_alias_id: The cloud provider alias ID of the agent to register.
                        In the Synapse API, this defaults to 'TSTALIASID'.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The registered agent matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/AgentRegistration.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request = {"awsAgentId": cloud_agent_id}
    if cloud_alias_id:
        request["awsAliasId"] = cloud_alias_id
    return await client.rest_put_async(
        uri="/agent/registration", body=json.dumps(request)
    )

get_agent async

get_agent(registration_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Gets information about an existing agent registration.

PARAMETER	DESCRIPTION
registration_id	The ID of the agent registration to get. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The requested agent registration matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/AgentRegistration.html

Source code in synapseclient/api/agent_services.py

async def get_agent(
    registration_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Gets information about an existing agent registration.

    Arguments:
        registration_id: The ID of the agent registration to get.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The requested agent registration matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/AgentRegistration.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(uri=f"/agent/registration/{registration_id}")

start_session async

start_session(access_level: str, agent_registration_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Starts a new chat session with an agent. Sends a request matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/CreateAgentSessionRequest.html

PARAMETER	DESCRIPTION
access_level	The access level of the agent. TYPE: str
agent_registration_id	The ID of the agent registration to start the session for. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Source code in synapseclient/api/agent_services.py

async def start_session(
    access_level: str,
    agent_registration_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Starts a new chat session with an agent.
    Sends a request matching
    <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/CreateAgentSessionRequest.html>

    Arguments:
        access_level: The access level of the agent.
        agent_registration_id: The ID of the agent registration to start the session for.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request = {
        "agentAccessLevel": access_level,
        "agentRegistrationId": agent_registration_id,
    }
    return await client.rest_post_async(uri="/agent/session", body=json.dumps(request))

get_session async

get_session(id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Gets information about an existing chat session.

PARAMETER	DESCRIPTION
id	The ID of the session to get. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The requested session matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/AgentSession.html

Source code in synapseclient/api/agent_services.py

async def get_session(
    id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Gets information about an existing chat session.

    Arguments:
        id: The ID of the session to get.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The requested session matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/AgentSession.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(uri=f"/agent/session/{id}")

update_session async

update_session(id: str, access_level: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Updates the access level for a chat session. Sends a request matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/UpdateAgentSessionRequest.html

PARAMETER	DESCRIPTION
id	The ID of the session to update. TYPE: str
access_level	The access level of the agent. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Source code in synapseclient/api/agent_services.py

async def update_session(
    id: str,
    access_level: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Updates the access level for a chat session.
    Sends a request matching
    <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/UpdateAgentSessionRequest.html>

    Arguments:
        id: The ID of the session to update.
        access_level: The access level of the agent.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request = {
        "sessionId": id,
        "agentAccessLevel": access_level,
    }
    return await client.rest_put_async(
        uri=f"/agent/session/{id}", body=json.dumps(request)
    )

get_trace async

get_trace(prompt_id: str, *, newer_than: Optional[int] = None, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Gets the trace of a prompt. Sends a request matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/TraceEventsRequest.html

PARAMETER	DESCRIPTION
prompt_id	The token of the prompt to get the trace for. TYPE: str
newer_than	The timestamp to get trace results newer than. Defaults to None (all results). Timestamps should be in milliseconds since the epoch per the API documentation. https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/TraceEvent.html TYPE: Optional[int] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The trace matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/TraceEventsResponse.html

Source code in synapseclient/api/agent_services.py

async def get_trace(
    prompt_id: str,
    *,
    newer_than: Optional[int] = None,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Gets the trace of a prompt.
    Sends a request matching
    <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/TraceEventsRequest.html>

    Arguments:
        prompt_id: The token of the prompt to get the trace for.
        newer_than: The timestamp to get trace results newer than. Defaults to None (all results).
                    Timestamps should be in milliseconds since the epoch per the API documentation.
                    https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/TraceEvent.html
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The trace matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/agent/TraceEventsResponse.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request = {
        "jobId": prompt_id,
        "newerThanTimestamp": newer_than,
    }
    return await client.rest_post_async(
        uri=f"/agent/chat/trace/{prompt_id}", body=json.dumps(request)
    )

synapseclient.api.annotations

The purpose of this module is to provide any functions that are needed to interact with annotations that are not cleanly provided by the synapseclient library.

Classes

Functions

set_annotations

set_annotations(annotations: Annotations, *, synapse_client: Optional[Synapse] = None)

Call to synapse and set the annotations for the given input.

PARAMETER	DESCRIPTION
annotations	The annotations to set. This is expected to have the id, etag, and annotations filled in. TYPE: Annotations
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Returns: The annotations set in Synapse.

Source code in synapseclient/api/annotations.py

def set_annotations(
    annotations: "Annotations",
    *,
    synapse_client: Optional["Synapse"] = None,
):
    """Call to synapse and set the annotations for the given input.

    Arguments:
        annotations: The annotations to set. This is expected to have the id, etag,
            and annotations filled in.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns: The annotations set in Synapse.
    """
    annotations_dict = asdict(annotations)

    synapse_annotations = _convert_to_annotations_list(
        annotations_dict["annotations"] or {}
    )
    from synapseclient import Synapse

    return Synapse.get_client(synapse_client=synapse_client).restPUT(
        f"/entity/{annotations.id}/annotations2",
        body=json.dumps(
            {
                "id": annotations.id,
                "etag": annotations.etag,
                "annotations": synapse_annotations,
            }
        ),
    )

set_annotations_async async

set_annotations_async(annotations: Annotations, *, synapse_client: Optional[Synapse] = None)

Call to synapse and set the annotations for the given input.

PARAMETER	DESCRIPTION
annotations	The annotations to set. This is expected to have the id, etag, and annotations filled in. TYPE: Annotations
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Returns: The annotations set in Synapse.

Source code in synapseclient/api/annotations.py

async def set_annotations_async(
    annotations: "Annotations",
    *,
    synapse_client: Optional["Synapse"] = None,
):
    """Call to synapse and set the annotations for the given input.

    Arguments:
        annotations: The annotations to set. This is expected to have the id, etag,
            and annotations filled in.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns: The annotations set in Synapse.
    """
    annotations_dict = asdict(annotations)

    synapse_annotations = _convert_to_annotations_list(
        annotations_dict["annotations"] or {}
    )
    from synapseclient import Synapse

    return await Synapse.get_client(synapse_client=synapse_client).rest_put_async(
        f"/entity/{annotations.id}/annotations2",
        body=json.dumps(
            {
                "id": annotations.id,
                "etag": annotations.etag,
                "annotations": synapse_annotations,
            }
        ),
    )

synapseclient.api.api_client

Classes

Functions

rest_post_paginated_async async

rest_post_paginated_async(uri: str, body: Optional[Dict[str, Any]] = None, endpoint: Optional[str] = None, headers: Optional[Headers] = None, retry_policy: Optional[Dict[str, Any]] = None, requests_session_async_synapse: Optional[AsyncClient] = None, *, synapse_client: Optional[Synapse] = None, **kwargs) -> AsyncGenerator[Dict[str, str], None]

Asynchronously yield items from a paginated POST endpoint.

PARAMETER	DESCRIPTION
uri	Endpoint URI for the POST request. TYPE: str
body	Request payload dictionary. TYPE: Optional[Dict[str, Any]] DEFAULT: None
endpoint	Optional server endpoint override. TYPE: Optional[str] DEFAULT: None
headers	Optional HTTP headers. TYPE: Optional[Headers] DEFAULT: None
retry_policy	Optional retry settings. TYPE: Optional[Dict[str, Any]] DEFAULT: None
requests_session_async_synapse	Optional async HTTPX client session. TYPE: Optional[AsyncClient] DEFAULT: None
kwargs	Additional keyword arguments for the request. DEFAULT: {}
synapse_client	Optional Synapse client instance for authentication. TYPE: Optional[Synapse] DEFAULT: None

Yields: Individual items from each page of the response.

Source code in synapseclient/api/api_client.py

async def rest_post_paginated_async(
    uri: str,
    body: Optional[Dict[str, Any]] = None,
    endpoint: Optional[str] = None,
    headers: Optional[httpx.Headers] = None,
    retry_policy: Optional[Dict[str, Any]] = None,
    requests_session_async_synapse: Optional[httpx.AsyncClient] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
    **kwargs,
) -> AsyncGenerator[Dict[str, str], None]:
    """
    Asynchronously yield items from a paginated POST endpoint.

    Arguments:
        uri: Endpoint URI for the POST request.
        body: Request payload dictionary.
        endpoint: Optional server endpoint override.
        headers: Optional HTTP headers.
        retry_policy: Optional retry settings.
        requests_session_async_synapse: Optional async HTTPX client session.
        kwargs: Additional keyword arguments for the request.
        synapse_client: Optional Synapse client instance for authentication.
    Yields:
        Individual items from each page of the response.
    """
    from synapseclient import Synapse

    if not retry_policy:
        retry_policy = {}

    client = Synapse.get_client(synapse_client=synapse_client)
    next_page_token = None
    while True:
        if next_page_token is not None:
            body = body or {}
            body["nextPageToken"] = next_page_token
        response = await client.rest_post_async(
            uri=uri,
            body=json.dumps(body),
            endpoint=endpoint,
            headers=headers,
            retry_policy=retry_policy,
            requests_session_async_synapse=requests_session_async_synapse,
            **kwargs,
        )
        next_page_token = response.get("nextPageToken")
        for item in response.get("page", []):
            yield item
        if next_page_token is None:
            break

rest_get_paginated_async async

rest_get_paginated_async(uri: str, limit: int = 20, offset: int = 0, endpoint: Optional[str] = None, headers: Optional[Headers] = None, retry_policy: Optional[Dict[str, Any]] = None, requests_session_async_synapse: Optional[AsyncClient] = None, *, synapse_client: Optional[Synapse] = None, **kwargs) -> AsyncGenerator[Dict[str, str], None]

Asynchronously yield items from a paginated GET endpoint.

PARAMETER	DESCRIPTION
uri	Endpoint URI for the GET request. TYPE: str
limit	How many records should be returned per request TYPE: int DEFAULT: 20
offset	At what record offset from the first should iteration start TYPE: int DEFAULT: 0
endpoint	Optional server endpoint override. TYPE: Optional[str] DEFAULT: None
headers	Optional HTTP headers. TYPE: Optional[Headers] DEFAULT: None
retry_policy	Optional retry settings. TYPE: Optional[Dict[str, Any]] DEFAULT: None
requests_session_async_synapse	Optional async HTTPX client session. TYPE: Optional[AsyncClient] DEFAULT: None
kwargs	Additional keyword arguments for the request. DEFAULT: {}
synapse_client	Optional Synapse client instance for authentication. TYPE: Optional[Synapse] DEFAULT: None

Yields: Individual items from each page of the response.

The limit parameter is set at 20 by default. Using a larger limit results in fewer calls to the service, but if responses are large enough to be a burden on the service they may be truncated.

Source code in synapseclient/api/api_client.py

async def rest_get_paginated_async(
    uri: str,
    limit: int = 20,
    offset: int = 0,
    endpoint: Optional[str] = None,
    headers: Optional[httpx.Headers] = None,
    retry_policy: Optional[Dict[str, Any]] = None,
    requests_session_async_synapse: Optional[httpx.AsyncClient] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
    **kwargs,
) -> AsyncGenerator[Dict[str, str], None]:
    """
    Asynchronously yield items from a paginated GET endpoint.

    Arguments:
        uri: Endpoint URI for the GET request.
        limit: How many records should be returned per request
        offset: At what record offset from the first should iteration start
        endpoint: Optional server endpoint override.
        headers: Optional HTTP headers.
        retry_policy: Optional retry settings.
        requests_session_async_synapse: Optional async HTTPX client session.
        kwargs: Additional keyword arguments for the request.
        synapse_client: Optional Synapse client instance for authentication.
    Yields:
        Individual items from each page of the response.

    The limit parameter is set at 20 by default. Using a larger limit results in fewer calls to the service, but if
    responses are large enough to be a burden on the service they may be truncated.
    """
    from synapseclient import Synapse
    from synapseclient.core import utils

    if not retry_policy:
        retry_policy = {}

    client = Synapse.get_client(synapse_client=synapse_client)
    prev_num_results = sys.maxsize
    while prev_num_results > 0:
        paginated_uri = utils._limit_and_offset(uri, limit=limit, offset=offset)
        response = await client.rest_get_async(
            uri=paginated_uri,
            endpoint=endpoint,
            headers=headers,
            retry_policy=retry_policy,
            requests_session_async_synapse=requests_session_async_synapse,
            **kwargs,
        )
        results = response["results"] if "results" in response else response["children"]
        prev_num_results = len(results)

        for result in results:
            offset += 1
            yield result

synapseclient.api.configuration_services

This module is responsible for exposing access to any configuration either through file, environment variables, or other means.

Functions

get_config_file cached

get_config_file(config_path: str) -> RawConfigParser

Retrieves the client configuration information.

PARAMETER	DESCRIPTION
config_path	Path to configuration file on local file system TYPE: str

RETURNS	DESCRIPTION
RawConfigParser	A RawConfigParser populated with properties from the user's configuration file.

Source code in synapseclient/api/configuration_services.py

@functools.lru_cache()
def get_config_file(config_path: str) -> configparser.RawConfigParser:
    """
    Retrieves the client configuration information.

    Arguments:
        config_path:  Path to configuration file on local file system

    Returns:
        A RawConfigParser populated with properties from the user's configuration file.
    """

    try:
        config = configparser.RawConfigParser()
        config.read(config_path)  # Does not fail if the file does not exist
        return config
    except configparser.Error as ex:
        raise ValueError(f"Error parsing Synapse config file: {config_path}") from ex

get_config_section_dict

get_config_section_dict(section_name: str, config_path: str) -> Dict[str, str]

Get a profile section in the configuration file with the section name.

PARAMETER	DESCRIPTION
section_name	The name of the profile section in the configuration file TYPE: str
config_path	Path to configuration file on local file system TYPE: str

RETURNS	DESCRIPTION
Dict[str, str]	A dictionary containing the configuration profile section content. If the
Dict[str, str]	section does not exist, an empty dictionary is returned.

Source code in synapseclient/api/configuration_services.py

def get_config_section_dict(
    section_name: str,
    config_path: str,
) -> Dict[str, str]:
    """
    Get a profile section in the configuration file with the section name.

    Arguments:
        section_name: The name of the profile section in the configuration file
        config_path:  Path to configuration file on local file system

    Returns:
        A dictionary containing the configuration profile section content. If the
        section does not exist, an empty dictionary is returned.
    """
    config = get_config_file(config_path)
    try:
        return dict(config.items(section_name))
    except configparser.NoSectionError:
        # section not present
        return {}

get_client_authenticated_s3_profile

get_client_authenticated_s3_profile(endpoint: str, bucket: str, config_path: str) -> Dict[str, str]

Get the authenticated S3 profile from the configuration file.

PARAMETER	DESCRIPTION
endpoint	The location of the target service TYPE: str
bucket	AWS S3 bucket name TYPE: str
config_path	Path to configuration file on local file system TYPE: str

RETURNS	DESCRIPTION
Dict[str, str]	The authenticated S3 profile

Source code in synapseclient/api/configuration_services.py

def get_client_authenticated_s3_profile(
    endpoint: str,
    bucket: str,
    config_path: str,
) -> Dict[str, str]:
    """
    Get the authenticated S3 profile from the configuration file.

    Arguments:
        endpoint: The location of the target service
        bucket:   AWS S3 bucket name
        config_path:  Path to configuration file on local file system

    Returns:
        The authenticated S3 profile
    """

    config_section = urllib.parse.urljoin(base=endpoint, url=bucket)
    return get_config_section_dict(
        section_name=config_section, config_path=config_path
    ).get("profile_name", "default")

get_config_authentication

get_config_authentication(config_path: str, profile: str = 'default') -> Dict[str, str]

Get the authentication section of the configuration file.

PARAMETER	DESCRIPTION
config_path	Path to configuration file on local file system TYPE: str
profile	The profile name to retrieve credentials for. Defaults to "default". TYPE: str DEFAULT: 'default'

RETURNS	DESCRIPTION
Dict[str, str]	The authentication section of the configuration file

Source code in synapseclient/api/configuration_services.py

def get_config_authentication(
    config_path: str, profile: str = "default"
) -> Dict[str, str]:
    """
    Get the authentication section of the configuration file.

    Arguments:
        config_path:  Path to configuration file on local file system
        profile (str, optional): The profile name to retrieve credentials for. Defaults to "default".

    Returns:
        The authentication section of the configuration file
    """

    section = profile if profile == "default" else f"profile {profile}"

    section_for_profile = get_config_section_dict(
        section_name=section,
        config_path=config_path,
    )

    if section_for_profile:
        return {
            "section_name": profile,
            **section_for_profile,
        }

    return {
        "section_name": config_file_constants.AUTHENTICATION_SECTION_NAME,
        **get_config_section_dict(
            section_name=config_file_constants.AUTHENTICATION_SECTION_NAME,
            config_path=config_path,
        ),
    }

get_transfer_config

get_transfer_config(config_path: str) -> Dict[str, str]

Get the transfer profile from the configuration file.

PARAMETER	DESCRIPTION
config_path	Path to configuration file on local file system TYPE: str

RAISES	DESCRIPTION
ValueError	Invalid max_threads value. Should be equal or less than 16.
ValueError	Invalid use_boto_sts value. Should be true or false.

RETURNS	DESCRIPTION
Dict[str, str]	The transfer profile

Source code in synapseclient/api/configuration_services.py

def get_transfer_config(
    config_path: str,
) -> Dict[str, str]:
    """
    Get the transfer profile from the configuration file.

    Arguments:
        config_path:  Path to configuration file on local file system

    Raises:
        ValueError: Invalid max_threads value. Should be equal or less than 16.
        ValueError: Invalid use_boto_sts value. Should be true or false.

    Returns:
        The transfer profile
    """
    # defaults
    transfer_config = {"max_threads": DEFAULT_NUM_THREADS, "use_boto_sts": False}

    for k, v in get_config_section_dict(
        section_name="transfer", config_path=config_path
    ).items():
        if v:
            if k == "max_threads":
                try:
                    transfer_config["max_threads"] = int(v)
                except ValueError as cause:
                    raise ValueError(
                        f"Invalid transfer.max_threads config setting {v}"
                    ) from cause

            elif k == "use_boto_sts":
                lower_v = v.lower()
                if lower_v not in ("true", "false"):
                    raise ValueError(
                        f"Invalid transfer.use_boto_sts config setting {v}"
                    )

                transfer_config["use_boto_sts"] = "true" == lower_v

    return transfer_config

synapseclient.api.entity_bundle_services_v2

This module is responsible for exposing the services defined at: https://rest-docs.synapse.org/rest/#org.sagebionetworks.repo.web.controller.EntityBundleV2Controller

Classes

Functions

get_entity_id_bundle2 async

get_entity_id_bundle2(entity_id: str, request: Optional[Dict[str, bool]] = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

PARAMETER	DESCRIPTION
entity_id	The ID of the entity to which the bundle belongs TYPE: str
request	The request for the bundle matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundleRequest.html. If not passed in or None, the default request will be used: includeEntity: True includeAnnotations: True includeFileHandles: True includeRestrictionInformation: True TYPE: Optional[Dict[str, bool]] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The requested entity bundle matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundle.html

Source code in synapseclient/api/entity_bundle_services_v2.py

async def get_entity_id_bundle2(
    entity_id: str,
    request: Optional[Dict[str, bool]] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Arguments:
        entity_id: The ID of the entity to which the bundle belongs
        request: The request for the bundle matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundleRequest.html>.
            If not passed in or None, the default request will be used:

            - includeEntity: True
            - includeAnnotations: True
            - includeFileHandles: True
            - includeRestrictionInformation: True
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The requested entity bundle matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundle.html>
    """
    from synapseclient import Synapse

    if not request:
        request = {
            "includeEntity": True,
            "includeAnnotations": True,
            "includeFileHandles": True,
            "includeRestrictionInformation": True,
        }

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_post_async(
        uri=f"/entity/{entity_id}/bundle2",
        body=json.dumps(request),
    )

get_entity_id_version_bundle2 async

get_entity_id_version_bundle2(entity_id: str, version: int, request: Optional[Dict[str, bool]] = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

PARAMETER	DESCRIPTION
entity_id	The ID of the entity to which the bundle belongs TYPE: str
version	The version of the entity to which the bundle belongs TYPE: int
request	The request for the bundle matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundleRequest.html. If not passed in or None, the default request will be used: includeEntity: True includeAnnotations: True includeFileHandles: True includeRestrictionInformation: True TYPE: Optional[Dict[str, bool]] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The requested entity bundle matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundle.html

Source code in synapseclient/api/entity_bundle_services_v2.py

async def get_entity_id_version_bundle2(
    entity_id: str,
    version: int,
    request: Optional[Dict[str, bool]] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Arguments:
        entity_id: The ID of the entity to which the bundle belongs
        version: The version of the entity to which the bundle belongs
        request: The request for the bundle matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundleRequest.html>.
            If not passed in or None, the default request will be used:

            - includeEntity: True
            - includeAnnotations: True
            - includeFileHandles: True
            - includeRestrictionInformation: True
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The requested entity bundle matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundle.html>
    """
    from synapseclient import Synapse

    if not request:
        request = {
            "includeEntity": True,
            "includeAnnotations": True,
            "includeFileHandles": True,
            "includeRestrictionInformation": True,
        }
    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_post_async(
        uri=f"/entity/{entity_id}/version/{version}/bundle2",
        body=json.dumps(request),
    )

post_entity_bundle2_create async

post_entity_bundle2_create(request: Dict[str, Any], generated_by: Optional[str] = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

PARAMETER	DESCRIPTION
request	The request for the bundle matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundleCreate.html TYPE: Dict[str, Any]
generated_by	The ID of the activity to associate with the entity. TYPE: Optional[str] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The requested entity bundle matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundle.html

Source code in synapseclient/api/entity_bundle_services_v2.py

async def post_entity_bundle2_create(
    request: Dict[str, Any],
    generated_by: Optional[str] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Arguments:
        request: The request for the bundle matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundleCreate.html>
        generated_by: The ID of the activity to associate with the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The requested entity bundle matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundle.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_post_async(
        uri="/entity/bundle2/create"
        + (f"?generatedBy={generated_by}" if generated_by else ""),
        body=json.dumps(request),
    )

put_entity_id_bundle2 async

put_entity_id_bundle2(entity_id: str, request: Dict[str, Any], generated_by: Optional[str] = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

PARAMETER	DESCRIPTION
entity_id	The ID of the entity to which the bundle belongs. TYPE: str
request	The request for the bundle matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundleCreate.html TYPE: Dict[str, Any]
generated_by	The ID of the activity to associate with the entity. TYPE: Optional[str] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The requested entity bundle matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundle.html

Source code in synapseclient/api/entity_bundle_services_v2.py

async def put_entity_id_bundle2(
    entity_id: str,
    request: Dict[str, Any],
    generated_by: Optional[str] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Arguments:
        entity_id: The ID of the entity to which the bundle belongs.
        request: The request for the bundle matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundleCreate.html>
        generated_by: The ID of the activity to associate with the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The requested entity bundle matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundle.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_put_async(
        uri=f"/entity/{entity_id}/bundle2"
        + (f"?generatedBy={generated_by}" if generated_by else ""),
        body=json.dumps(request),
    )

synapseclient.api.entity_factory

Factory type functions to create and retrieve entities from Synapse

Classes

Functions

get_from_entity_factory async

get_from_entity_factory(synapse_id_or_path: str, version: int = None, if_collision: str = 'keep.both', limit_search: str = None, md5: str = None, download_file: bool = True, download_location: str = None, follow_link: bool = False, entity_to_update: Union[Project, File, Folder, Table, Dataset, EntityView] = None, *, synapse_client: Optional[Synapse] = None) -> Union[Project, File, Folder]

Factory type function to retrieve an entity from Synapse. Optionally you may also pass in entity_to_update if you want to update the fields on the existing entity instead of creating a new instance.

PARAMETER	DESCRIPTION
synapse_id_or_path	The Synapse ID or file path of the entity to retrieve. TYPE: str
version	The version number of the entity to retrieve. TYPE: int DEFAULT: None
if_collision	Determines how to handle file collisions. May be: - `overwrite.local` - `keep.local` - `keep.both` TYPE: str DEFAULT: 'keep.both'
limit_search	Limit the search to a specific project or folder. Only used if synapse_id_or_path is a path. TYPE: str DEFAULT: None
md5	The MD5 of the file to retrieve. If not passed in, the MD5 will be calculated. Only used if synapse_id_or_path is a path. TYPE: str DEFAULT: None
download_file	Whether associated files should be downloaded. TYPE: bool DEFAULT: True
download_location	The directory to download the file to. TYPE: str DEFAULT: None
follow_link	Whether to follow a link to its target. This will only do a single hop to the target of the link. TYPE: bool DEFAULT: False
entity_to_update	An existing entity class instance to update with data from Synapse. TYPE: Union[Project, File, Folder, Table, Dataset, EntityView] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None
Example	Using this function Download file into cache from synapseclient import Synapse from synapseclient.api import get_from_entity_factory syn = Synapse() syn.login() entity = await get_from_entity_factory(synapse_id_or_path='syn1906479') print(entity.name) print(entity.path) Download file into current working directory from synapseclient import Synapse from synapseclient.api import get_from_entity_factory syn = Synapse() syn.login() entity = await get_from_entity_factory(synapse_id_or_path='syn1906479', download_location='.') print(entity.name) print(entity.path)

RAISES	DESCRIPTION
SynapseFileNotFoundError	If the id is not a synapse ID and it is not a valid file path.

Source code in synapseclient/api/entity_factory.py

async def get_from_entity_factory(
    synapse_id_or_path: str,
    version: int = None,
    if_collision: str = "keep.both",
    limit_search: str = None,
    md5: str = None,
    download_file: bool = True,
    download_location: str = None,
    follow_link: bool = False,
    entity_to_update: Union[
        "Project", "File", "Folder", "Table", "Dataset", "EntityView"
    ] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Union["Project", "File", "Folder"]:
    """
    Factory type function to retrieve an entity from Synapse. Optionally you may also
    pass in `entity_to_update` if you want to update the fields on the existing entity
    instead of creating a new instance.

    Arguments:
        synapse_id_or_path: The Synapse ID or file path of the entity to retrieve.
        version:            The version number of the entity to retrieve.
        if_collision: Determines how to handle file collisions. May be:

                - `overwrite.local`
                - `keep.local`
                - `keep.both`
        limit_search:       Limit the search to a specific project or folder. Only used
            if `synapse_id_or_path` is a path.
        md5: The MD5 of the file to retrieve. If not passed in, the MD5 will be
            calculated. Only used if `synapse_id_or_path` is a path.
        download_file: Whether associated files should be downloaded.
        download_location: The directory to download the file to.
        follow_link: Whether to follow a link to its target. This will only do a single
            hop to the target of the link.
        entity_to_update: An existing entity class instance to update with data from
            Synapse.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Example: Using this function
            Download file into cache

                from synapseclient import Synapse
                from synapseclient.api import get_from_entity_factory

                syn = Synapse()
                syn.login()

                entity = await get_from_entity_factory(synapse_id_or_path='syn1906479')
                print(entity.name)
                print(entity.path)

            Download file into current working directory

                from synapseclient import Synapse
                from synapseclient.api import get_from_entity_factory

                syn = Synapse()
                syn.login()

                entity = await get_from_entity_factory(synapse_id_or_path='syn1906479', download_location='.')
                print(entity.name)
                print(entity.path)

    Raises:
        SynapseFileNotFoundError: If the id is not a synapse ID and it is not a valid
            file path.
    """

    # If entity is a local file determine the corresponding synapse entity
    if isinstance(synapse_id_or_path, str) and os.path.isfile(synapse_id_or_path):
        bundle = await _search_for_file_by_md5(
            filepath=synapse_id_or_path,
            limit_search=limit_search,
            md5=md5,
            synapse_client=synapse_client,
        )
        download_file = False

    elif isinstance(synapse_id_or_path, str) and not utils.is_synapse_id_str(
        obj=synapse_id_or_path
    ):
        raise SynapseFileNotFoundError(
            (
                f"The parameter {synapse_id_or_path} is neither a local file path "
                " or a valid entity id"
            )
        )
    else:
        synid_and_version = utils.get_synid_and_version(obj=synapse_id_or_path)
        version = version if version is not None else synid_and_version[1]
        if version:
            bundle = await get_entity_id_version_bundle2(
                entity_id=synid_and_version[0],
                version=version,
                synapse_client=synapse_client,
            )
        else:
            bundle = await get_entity_id_bundle2(
                entity_id=synid_and_version[0], synapse_client=synapse_client
            )

    # Check and warn for unmet access requirements
    _check_entity_restrictions(
        bundle=bundle,
        synapse_id=bundle["entity"]["id"],
        download_file=download_file,
        synapse_client=synapse_client,
    )

    return_data = await _cast_into_class_type(
        entity_to_update=entity_to_update,
        entity_bundle=bundle,
        download_file=download_file,
        download_location=download_location,
        if_collision=if_collision,
        follow_link=follow_link,
        synapse_client=synapse_client,
    )
    trace.get_current_span().set_attributes(
        {
            "synapse.id": return_data.id,
            "synapse.type": return_data.__class__.__name__,
        }
    )
    return return_data

synapseclient.api.entity_services

This module is responsible for exposing the services defined at: https://rest-docs.synapse.org/rest/#org.sagebionetworks.repo.web.controller.EntityController

Classes

EntityHeader dataclass

JSON schema for EntityHeader POJO. This represents metadata about a Synapse entity.

ATTRIBUTE	DESCRIPTION
name	The name of the entity TYPE: Optional[str]
id	The id of the entity TYPE: Optional[str]
type	The type of the entity TYPE: Optional[str]
version_number	The version number of the entity TYPE: Optional[int]
version_label	The user defined version label of the entity TYPE: Optional[str]
is_latest_version	If this version is the latest version of the entity TYPE: Optional[bool]
benefactor_id	The ID of the entity that this Entity's ACL is inherited from TYPE: Optional[int]
created_on	The date this entity was created TYPE: Optional[str]
modified_on	The date this entity was last modified TYPE: Optional[str]
created_by	The ID of the user that created this entity TYPE: Optional[str]
modified_by	The ID of the user that last modified this entity TYPE: Optional[str]

Source code in synapseclient/api/entity_services.py

@dataclass
class EntityHeader:
    """
    JSON schema for EntityHeader POJO. This represents metadata about a Synapse entity.

    Attributes:
        name: The name of the entity
        id: The id of the entity
        type: The type of the entity
        version_number: The version number of the entity
        version_label: The user defined version label of the entity
        is_latest_version: If this version is the latest version of the entity
        benefactor_id: The ID of the entity that this Entity's ACL is inherited from
        created_on: The date this entity was created
        modified_on: The date this entity was last modified
        created_by: The ID of the user that created this entity
        modified_by: The ID of the user that last modified this entity
    """

    name: Optional[str] = None
    """The name of the entity"""

    id: Optional[str] = None
    """The id of the entity"""

    type: Optional[str] = None
    """The type of the entity"""

    version_number: Optional[int] = None
    """The version number of the entity"""

    version_label: Optional[str] = None
    """The user defined version label of the entity"""

    is_latest_version: Optional[bool] = None
    """If this version is the latest version of the entity"""

    benefactor_id: Optional[int] = None
    """The ID of the entity that this Entity's ACL is inherited from"""

    created_on: Optional[str] = None
    """The date this entity was created"""

    modified_on: Optional[str] = None
    """The date this entity was last modified"""

    created_by: Optional[str] = None
    """The ID of the user that created this entity"""

    modified_by: Optional[str] = None
    """The ID of the user that last modified this entity"""

    def fill_from_dict(self, synapse_response: Dict[str, Any]) -> "EntityHeader":
        """Converts a response from the REST API into this dataclass."""
        self.name = synapse_response.get("name", None)
        self.id = synapse_response.get("id", None)
        self.type = synapse_response.get("type", None)
        self.version_number = synapse_response.get("versionNumber", None)
        self.version_label = synapse_response.get("versionLabel", None)
        self.is_latest_version = synapse_response.get("isLatestVersion", None)
        self.benefactor_id = synapse_response.get("benefactorId", None)
        self.created_on = synapse_response.get("createdOn", None)
        self.modified_on = synapse_response.get("modifiedOn", None)
        self.created_by = synapse_response.get("createdBy", None)
        self.modified_by = synapse_response.get("modifiedBy", None)
        return self

Attributes

name class-attribute instance-attribute

name: Optional[str] = None

The name of the entity

id class-attribute instance-attribute

id: Optional[str] = None

The id of the entity

type class-attribute instance-attribute

type: Optional[str] = None

The type of the entity

version_number class-attribute instance-attribute

version_number: Optional[int] = None

The version number of the entity

version_label class-attribute instance-attribute

version_label: Optional[str] = None

The user defined version label of the entity

is_latest_version class-attribute instance-attribute

is_latest_version: Optional[bool] = None

If this version is the latest version of the entity

benefactor_id class-attribute instance-attribute

benefactor_id: Optional[int] = None

The ID of the entity that this Entity's ACL is inherited from

created_on class-attribute instance-attribute

created_on: Optional[str] = None

The date this entity was created

modified_on class-attribute instance-attribute

modified_on: Optional[str] = None

The date this entity was last modified

created_by class-attribute instance-attribute

created_by: Optional[str] = None

The ID of the user that created this entity

modified_by class-attribute instance-attribute

modified_by: Optional[str] = None

The ID of the user that last modified this entity

Functions

fill_from_dict

fill_from_dict(synapse_response: Dict[str, Any]) -> EntityHeader

Converts a response from the REST API into this dataclass.

Source code in synapseclient/api/entity_services.py

def fill_from_dict(self, synapse_response: Dict[str, Any]) -> "EntityHeader":
    """Converts a response from the REST API into this dataclass."""
    self.name = synapse_response.get("name", None)
    self.id = synapse_response.get("id", None)
    self.type = synapse_response.get("type", None)
    self.version_number = synapse_response.get("versionNumber", None)
    self.version_label = synapse_response.get("versionLabel", None)
    self.is_latest_version = synapse_response.get("isLatestVersion", None)
    self.benefactor_id = synapse_response.get("benefactorId", None)
    self.created_on = synapse_response.get("createdOn", None)
    self.modified_on = synapse_response.get("modifiedOn", None)
    self.created_by = synapse_response.get("createdBy", None)
    self.modified_by = synapse_response.get("modifiedBy", None)
    return self

Functions

post_entity async

post_entity(request: Dict[str, Any], generated_by: Optional[str] = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

PARAMETER	DESCRIPTION
request	The request for the entity matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Entity.html TYPE: Dict[str, Any]
generated_by	The ID of the activity to associate with the entity. TYPE: Optional[str] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The requested entity matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Entity.html

Source code in synapseclient/api/entity_services.py

async def post_entity(
    request: Dict[str, Any],
    generated_by: Optional[str] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Arguments:
        request: The request for the entity matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Entity.html>
        generated_by: The ID of the activity to associate with the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The requested entity matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Entity.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    params = {}
    if generated_by:
        params["generatedBy"] = generated_by
    return await client.rest_post_async(
        uri="/entity", body=json.dumps(request), params=params
    )

put_entity async

put_entity(entity_id: str, request: Dict[str, Any], new_version: bool = False, generated_by: Optional[str] = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

PARAMETER	DESCRIPTION
entity_id	The ID of the entity to update. TYPE: str
request	The request for the entity matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Entity.html TYPE: Dict[str, Any]
new_version	If true, a new version of the entity will be created. TYPE: bool DEFAULT: False
generated_by	The ID of the activity to associate with the entity. TYPE: Optional[str] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The requested entity bundle matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Entity.html

Source code in synapseclient/api/entity_services.py

async def put_entity(
    entity_id: str,
    request: Dict[str, Any],
    new_version: bool = False,
    generated_by: Optional[str] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Arguments:
        entity_id: The ID of the entity to update.
        request: The request for the entity matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Entity.html>
        new_version: If true, a new version of the entity will be created.
        generated_by: The ID of the activity to associate with the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The requested entity bundle matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Entity.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    params = {}
    if generated_by:
        params["generatedBy"] = generated_by
    if new_version:
        params["newVersion"] = "true"
    return await client.rest_put_async(
        uri=f"/entity/{entity_id}", body=json.dumps(request), params=params
    )

get_entity async

get_entity(entity_id: str, version_number: int = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The requested entity bundle matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Entity.html

Source code in synapseclient/api/entity_services.py

async def get_entity(
    entity_id: str,
    version_number: int = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Arguments:
        entity_id: The ID of the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The requested entity bundle matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Entity.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    if version_number:
        return await client.rest_get_async(
            uri=f"/entity/{entity_id}/version/{version_number}",
        )
    else:
        return await client.rest_get_async(
            uri=f"/entity/{entity_id}",
        )

get_upload_destination async

get_upload_destination(entity_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, int]]

https://rest-docs.synapse.org/rest/GET/entity/id/uploadDestination.html

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, int]]	The upload destination.
Dict[str, Union[str, int]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/UploadDestination.html

Source code in synapseclient/api/entity_services.py

@alru_cache(ttl=60)
async def get_upload_destination(
    entity_id: str, *, synapse_client: Optional["Synapse"] = None
) -> Dict[str, Union[str, int]]:
    """
    <https://rest-docs.synapse.org/rest/GET/entity/id/uploadDestination.html>

    Arguments:
        entity_id: The ID of the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The upload destination.
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/UploadDestination.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(
        uri=f"/entity/{entity_id}/uploadDestination",
        endpoint=client.fileHandleEndpoint,
    )

get_upload_destination_location async

get_upload_destination_location(entity_id: str, location: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, int]]

https://rest-docs.synapse.org/rest/GET/entity/id/uploadDestination/storageLocationId.html

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
location	A storage location ID of the upload destination. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, int]]	The upload destination.
Dict[str, Union[str, int]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/UploadDestination.html

Source code in synapseclient/api/entity_services.py

async def get_upload_destination_location(
    entity_id: str, location: str, *, synapse_client: Optional["Synapse"] = None
) -> Dict[str, Union[str, int]]:
    """
    <https://rest-docs.synapse.org/rest/GET/entity/id/uploadDestination/storageLocationId.html>

    Arguments:
        entity_id: The ID of the entity.
        location: A storage location ID of the upload destination.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The upload destination.
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/UploadDestination.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(
        uri=f"/entity/{entity_id}/uploadDestination/{location}",
        endpoint=client.fileHandleEndpoint,
    )

create_access_requirements_if_none async

create_access_requirements_if_none(entity_id: str, *, synapse_client: Optional[Synapse] = None) -> None

Checks to see if the given entity has access requirements. If not, then one is added

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Source code in synapseclient/api/entity_services.py

async def create_access_requirements_if_none(
    entity_id: str, *, synapse_client: Optional["Synapse"] = None
) -> None:
    """
    Checks to see if the given entity has access requirements. If not, then one is added

    Arguments:
        entity_id: The ID of the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    existing_restrictions = await client.rest_get_async(
        f"/entity/{entity_id}/accessRequirement?offset=0&limit=1"
    )
    if (
        existing_restrictions is None
        or not hasattr(existing_restrictions, "results")
        or len(existing_restrictions["results"]) == 0
    ):
        access_requirements = await client.rest_post_async(
            f"/entity/{entity_id}/lockAccessRequirement"
        )
        client.logger.info(
            "Created an access requirements request for "
            f"{entity_id}: {access_requirements['jiraKey']}. An email will be sent to "
            "the Synapse access control team to start the process of adding "
            "terms-of-use or review board approval for this entity."
        )

delete_entity_generated_by async

delete_entity_generated_by(entity_id: str, *, synapse_client: Optional[Synapse] = None) -> None

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Returns: None

Source code in synapseclient/api/entity_services.py

async def delete_entity_generated_by(
    entity_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> None:
    """
    Arguments:
        entity_id: The ID of the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns: None
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_delete_async(
        uri=f"/entity/{entity_id}/generatedBy",
    )

delete_entity async

delete_entity(entity_id: str, version_number: int = None, *, synapse_client: Optional[Synapse] = None) -> None

Deletes an entity from Synapse.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. This may include version syn123.0 or syn123. If the version is included in entity_id and version_number is also passed in, then the version in entity_id will be used. TYPE: str
version_number	The version number of the entity to delete. TYPE: int DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Delete the entity syn123:

This will delete all versions of the entity.

import asyncio
from synapseclient import Synapse
from synapseclient.api import delete_entity

syn = Synapse()
syn.login()


async def main():
    await delete_entity(entity_id="syn123")

asyncio.run(main())

Delete a specific version of the entity syn123:

This will delete version 3 of the entity.

import asyncio
from synapseclient import Synapse
from synapseclient.api import delete_entity

syn = Synapse()
syn.login()


async def main():
    await delete_entity(entity_id="syn123", version_number=3)

asyncio.run(main())

Returns: None

Source code in synapseclient/api/entity_services.py

async def delete_entity(
    entity_id: str,
    version_number: int = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> None:
    """
    Deletes an entity from Synapse.

    Arguments:
        entity_id: The ID of the entity. This may include version `syn123.0` or `syn123`.
            If the version is included in `entity_id` and `version_number` is also
            passed in, then the version in `entity_id` will be used.
        version_number: The version number of the entity to delete.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Example: Delete the entity `syn123`:
        This will delete all versions of the entity.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import delete_entity

        syn = Synapse()
        syn.login()


        async def main():
            await delete_entity(entity_id="syn123")

        asyncio.run(main())
        ```

    Example: Delete a specific version of the entity `syn123`:
        This will delete version `3` of the entity.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import delete_entity

        syn = Synapse()
        syn.login()


        async def main():
            await delete_entity(entity_id="syn123", version_number=3)

        asyncio.run(main())
        ```

    Returns: None
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    syn_id, syn_version = get_synid_and_version(entity_id)
    if not syn_version:
        syn_version = version_number

    if syn_version:
        return await client.rest_delete_async(
            uri=f"/entity/{syn_id}/version/{syn_version}",
        )
    else:
        return await client.rest_delete_async(
            uri=f"/entity/{syn_id}",
        )

delete_entity_acl async

delete_entity_acl(entity_id: str, *, synapse_client: Optional[Synapse] = None) -> None

Delete the Access Control List (ACL) for a given Entity.

By default, Entities such as FileEntity and Folder inherit their permission from their containing Project. For such Entities the Project is the Entity's 'benefactor'. This permission inheritance can be overridden by creating an ACL for the Entity. When this occurs the Entity becomes its own benefactor and all permission are determined by its own ACL.

If the ACL of an Entity is deleted, then its benefactor will automatically be set to its parent's benefactor. The ACL for a Project cannot be deleted.

Note: The caller must be granted ACCESS_TYPE.CHANGE_PERMISSIONS on the Entity to call this method.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity that should have its ACL deleted. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Delete the ACL for entity syn123:

This will delete the ACL for the entity, making it inherit permissions from its parent.

import asyncio
from synapseclient import Synapse
from synapseclient.api import delete_entity_acl

syn = Synapse()
syn.login()

async def main():
    await delete_entity_acl(entity_id="syn123")

asyncio.run(main())

Returns: None

Source code in synapseclient/api/entity_services.py

async def delete_entity_acl(
    entity_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> None:
    """
    Delete the Access Control List (ACL) for a given Entity.

    By default, Entities such as FileEntity and Folder inherit their permission from
    their containing Project. For such Entities the Project is the Entity's 'benefactor'.
    This permission inheritance can be overridden by creating an ACL for the Entity.
    When this occurs the Entity becomes its own benefactor and all permission are
    determined by its own ACL.

    If the ACL of an Entity is deleted, then its benefactor will automatically be set
    to its parent's benefactor. The ACL for a Project cannot be deleted.

    Note: The caller must be granted ACCESS_TYPE.CHANGE_PERMISSIONS on the Entity to
    call this method.

    Arguments:
        entity_id: The ID of the entity that should have its ACL deleted.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Example: Delete the ACL for entity `syn123`:
        This will delete the ACL for the entity, making it inherit permissions from
        its parent.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import delete_entity_acl

        syn = Synapse()
        syn.login()

        async def main():
            await delete_entity_acl(entity_id="syn123")

        asyncio.run(main())
        ```

    Returns: None
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_delete_async(
        uri=f"/entity/{entity_id}/acl",
    )

get_entity_acl async

get_entity_acl(entity_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]

Get the Access Control List (ACL) for an entity.

Note: If this method is called on an Entity that is inheriting its permission from another Entity a NOT_FOUND (404) response will be generated. The error response message will include the Entity's benefactor ID.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]	A dictionary of the Entity's ACL.
Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html

Source code in synapseclient/api/entity_services.py

async def get_entity_acl(
    entity_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]:
    """
    Get the Access Control List (ACL) for an entity.

    Note: If this method is called on an Entity that is inheriting its permission
    from another Entity a NOT_FOUND (404) response will be generated. The error
    response message will include the Entity's benefactor ID.

    Arguments:
        entity_id: The ID of the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        A dictionary of the Entity's ACL.
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(
        uri=f"/entity/{entity_id}/acl",
    )

get_entity_acl_with_benefactor async

get_entity_acl_with_benefactor(entity_id: str, check_benefactor: bool = True, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]

Get the effective Access Control List (ACL) for a Synapse Entity.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
check_benefactor	If True (default), check the benefactor for the entity to get the ACL. If False, only check the entity itself. This is useful for checking the ACL of an entity that has local sharing settings, but you want to check the ACL of the entity itself and not the benefactor it may inherit from. TYPE: bool DEFAULT: True
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]	A dictionary of the Entity's ACL.
https	//rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html TYPE: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]
Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]	If the entity does not have its own ACL and check_benefactor is False,
Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]	returns {"resourceAccess": []}.

Get ACL with benefactor checking

Get the effective ACL for entity syn123.

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_entity_acl_with_benefactor

syn = Synapse()
syn.login()

async def main():
    # Get ACL from benefactor (default behavior)
    acl = await get_entity_acl_with_benefactor(entity_id="syn123")
    print(f"ACL from benefactor: {acl}")

    # Get ACL from entity only
    acl = await get_entity_acl_with_benefactor(
        entity_id="syn123",
        check_benefactor=False
    )
    print(f"ACL from entity only: {acl}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def get_entity_acl_with_benefactor(
    entity_id: str,
    check_benefactor: bool = True,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]:
    """
    Get the effective Access Control List (ACL) for a Synapse Entity.

    Arguments:
        entity_id: The ID of the entity.
        check_benefactor: If True (default), check the benefactor for the entity
                         to get the ACL. If False, only check the entity itself.
                         This is useful for checking the ACL of an entity that has local sharing
                         settings, but you want to check the ACL of the entity itself and not
                         the benefactor it may inherit from.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        A dictionary of the Entity's ACL.
        https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html
        If the entity does not have its own ACL and check_benefactor is False,
        returns {"resourceAccess": []}.

    Example: Get ACL with benefactor checking
        Get the effective ACL for entity `syn123`.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_entity_acl_with_benefactor

        syn = Synapse()
        syn.login()

        async def main():
            # Get ACL from benefactor (default behavior)
            acl = await get_entity_acl_with_benefactor(entity_id="syn123")
            print(f"ACL from benefactor: {acl}")

            # Get ACL from entity only
            acl = await get_entity_acl_with_benefactor(
                entity_id="syn123",
                check_benefactor=False
            )
            print(f"ACL from entity only: {acl}")

        asyncio.run(main())
        ```
    """
    if check_benefactor:
        # Get the ACL from the benefactor (which may be the entity itself)
        benefactor = await get_entity_benefactor(
            entity_id=entity_id, synapse_client=synapse_client
        )
        return await get_entity_acl(
            entity_id=benefactor.id, synapse_client=synapse_client
        )
    else:
        try:
            return await get_entity_acl(
                entity_id=entity_id, synapse_client=synapse_client
            )
        except SynapseHTTPError as e:
            # If entity doesn't have its own ACL and check_benefactor is False,
            # return empty ACL structure indicating no local permissions
            if (
                "The requested ACL does not exist. This entity inherits its permissions from:"
                in str(e)
            ):
                return {"resourceAccess": []}
            raise e

put_entity_acl async

put_entity_acl(entity_id: str, acl: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]], *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]

Update the Access Control List (ACL) for an entity.

API Matches https://rest-docs.synapse.org/rest/PUT/entity/id/acl.html.

Note: The caller must be granted CHANGE_PERMISSIONS on the Entity to call this method.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
acl	The ACL to set for the entity. TYPE: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]	The updated ACL matching
https	//rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html TYPE: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]

Update ACL for an entity

Update the ACL for entity syn123.

import asyncio
from synapseclient import Synapse
from synapseclient.api import put_entity_acl

syn = Synapse()
syn.login()

async def main():
    acl = {
        "id": "syn123",
        "etag": "12345",
        "resourceAccess": [
            {
                "principalId": 12345,
                "accessType": ["READ", "DOWNLOAD"]
            }
        ]
    }
    updated_acl = await put_entity_acl(entity_id="syn123", acl=acl)
    print(f"Updated ACL: {updated_acl}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def put_entity_acl(
    entity_id: str,
    acl: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]:
    """
    Update the Access Control List (ACL) for an entity.

    API Matches <https://rest-docs.synapse.org/rest/PUT/entity/id/acl.html>.

    Note: The caller must be granted `CHANGE_PERMISSIONS` on the Entity to call this method.

    Arguments:
        entity_id: The ID of the entity.
        acl: The ACL to set for the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The updated ACL matching
        https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html

    Example: Update ACL for an entity
        Update the ACL for entity `syn123`.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import put_entity_acl

        syn = Synapse()
        syn.login()

        async def main():
            acl = {
                "id": "syn123",
                "etag": "12345",
                "resourceAccess": [
                    {
                        "principalId": 12345,
                        "accessType": ["READ", "DOWNLOAD"]
                    }
                ]
            }
            updated_acl = await put_entity_acl(entity_id="syn123", acl=acl)
            print(f"Updated ACL: {updated_acl}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_put_async(
        uri=f"/entity/{entity_id}/acl",
        body=json.dumps(acl),
    )

post_entity_acl async

post_entity_acl(entity_id: str, acl: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]], *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]

Create a new Access Control List (ACL) for an entity.

API Matches https://rest-docs.synapse.org/rest/POST/entity/id/acl.html.

Note: The caller must be granted CHANGE_PERMISSIONS on the Entity to call this method.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
acl	The ACL to create for the entity. TYPE: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]	The created ACL matching
Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html.

Create ACL for an entity

Create a new ACL for entity syn123.

import asyncio
from synapseclient import Synapse
from synapseclient.api import post_entity_acl

syn = Synapse()
syn.login()

async def main():
    acl = {
        "id": "syn123",
        "etag": "12345",
        "resourceAccess": [
            {
                "principalId": 12345,
                "accessType": ["READ", "DOWNLOAD"]
            }
        ]
    }
    created_acl = await post_entity_acl(entity_id="syn123", acl=acl)
    print(f"Created ACL: {created_acl}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def post_entity_acl(
    entity_id: str,
    acl: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]:
    """
    Create a new Access Control List (ACL) for an entity.

    API Matches <https://rest-docs.synapse.org/rest/POST/entity/id/acl.html>.

    Note: The caller must be granted `CHANGE_PERMISSIONS` on the Entity to call this method.

    Arguments:
        entity_id: The ID of the entity.
        acl: The ACL to create for the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The created ACL matching
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html>.

    Example: Create ACL for an entity
        Create a new ACL for entity `syn123`.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import post_entity_acl

        syn = Synapse()
        syn.login()

        async def main():
            acl = {
                "id": "syn123",
                "etag": "12345",
                "resourceAccess": [
                    {
                        "principalId": 12345,
                        "accessType": ["READ", "DOWNLOAD"]
                    }
                ]
            }
            created_acl = await post_entity_acl(entity_id="syn123", acl=acl)
            print(f"Created ACL: {created_acl}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_post_async(
        uri=f"/entity/{entity_id}/acl",
        body=json.dumps(acl),
    )

get_entity_permissions async

get_entity_permissions(entity_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, List[str], bool]]

Get the permissions that the caller has on an Entity.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, List[str], bool]]	A dictionary containing the permissions that the caller has on the entity.
Dict[str, Union[str, List[str], bool]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/auth/UserEntityPermissions.html

Get permissions for an entity

Get the permissions that the caller has on entity syn123.

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_entity_permissions

syn = Synapse()
syn.login()

async def main():
    permissions = await get_entity_permissions(entity_id="syn123")
    print(f"Permissions: {permissions}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def get_entity_permissions(
    entity_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, List[str], bool]]:
    """
    Get the permissions that the caller has on an Entity.

    Arguments:
        entity_id: The ID of the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        A dictionary containing the permissions that the caller has on the entity.
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/auth/UserEntityPermissions.html>

    Example: Get permissions for an entity
        Get the permissions that the caller has on entity `syn123`.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_entity_permissions

        syn = Synapse()
        syn.login()

        async def main():
            permissions = await get_entity_permissions(entity_id="syn123")
            print(f"Permissions: {permissions}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(
        uri=f"/entity/{entity_id}/permissions",
    )

get_entity_benefactor async

get_entity_benefactor(entity_id: str, *, synapse_client: Optional[Synapse] = None) -> EntityHeader

Get an Entity's benefactor. An Entity gets its ACL from its benefactor.

Implements: https://rest-docs.synapse.org/rest/GET/entity/id/benefactor.html

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
EntityHeader	The EntityHeader of the entity's benefactor (the entity from which it inherits its ACL).

Get the benefactor of an entity

Get the benefactor entity header for entity syn123.

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_entity_benefactor

syn = Synapse()
syn.login()

async def main():
    benefactor = await get_entity_benefactor(entity_id="syn123")
    print(f"Entity benefactor: {benefactor.name} (ID: {benefactor.id})")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def get_entity_benefactor(
    entity_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> EntityHeader:
    """
    Get an Entity's benefactor. An Entity gets its ACL from its benefactor.

    Implements:
    <https://rest-docs.synapse.org/rest/GET/entity/id/benefactor.html>

    Arguments:
        entity_id: The ID of the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The EntityHeader of the entity's benefactor (the entity from which it inherits its ACL).

    Example: Get the benefactor of an entity
        Get the benefactor entity header for entity `syn123`.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_entity_benefactor

        syn = Synapse()
        syn.login()

        async def main():
            benefactor = await get_entity_benefactor(entity_id="syn123")
            print(f"Entity benefactor: {benefactor.name} (ID: {benefactor.id})")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    response = await client.rest_get_async(
        uri=f"/entity/{entity_id}/benefactor",
    )

    entity_header = EntityHeader()
    return entity_header.fill_from_dict(response)

get_entity_path async

get_entity_path(entity_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, List[Dict[str, Union[str, int, bool]]]]

Implements: https://rest-docs.synapse.org/rest/GET/entity/id/path.html

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, List[Dict[str, Union[str, int, bool]]]]	Entity paths matching:
Dict[str, List[Dict[str, Union[str, int, bool]]]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/EntityPath.html

Source code in synapseclient/api/entity_services.py

async def get_entity_path(
    entity_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, List[Dict[str, Union[str, int, bool]]]]:
    """
    Implements:
    <https://rest-docs.synapse.org/rest/GET/entity/id/path.html>

    Arguments:
        entity_id: The ID of the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        Entity paths matching:
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/EntityPath.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(
        uri=f"/entity/{entity_id}/path",
    )

get_entity_type async

get_entity_type(entity_id: str, *, synapse_client: Optional[Synapse] = None) -> EntityHeader

Get the EntityHeader of an Entity given its ID. The EntityHeader is a light weight object with basic information about an Entity includes its type.

Implements: https://rest-docs.synapse.org/rest/GET/entity/id/type.html

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
EntityHeader	EntityHeader object containing basic information about the entity.

Get entity type information

Get the EntityHeader for entity syn123.

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_entity_type

syn = Synapse()
syn.login()

async def main():
    entity_header = await get_entity_type(entity_id="syn123")
    print(f"Entity type: {entity_header.type}")
    print(f"Entity name: {entity_header.name}")
    print(f"Entity ID: {entity_header.id}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def get_entity_type(
    entity_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> EntityHeader:
    """
    Get the EntityHeader of an Entity given its ID. The EntityHeader is a light weight
    object with basic information about an Entity includes its type.

    Implements:
    <https://rest-docs.synapse.org/rest/GET/entity/id/type.html>

    Arguments:
        entity_id: The ID of the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        EntityHeader object containing basic information about the entity.

    Example: Get entity type information
        Get the EntityHeader for entity `syn123`.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_entity_type

        syn = Synapse()
        syn.login()

        async def main():
            entity_header = await get_entity_type(entity_id="syn123")
            print(f"Entity type: {entity_header.type}")
            print(f"Entity name: {entity_header.name}")
            print(f"Entity ID: {entity_header.id}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    response = await client.rest_get_async(
        uri=f"/entity/{entity_id}/type",
    )

    entity_header = EntityHeader()
    return entity_header.fill_from_dict(response)

get_entities_by_md5 async

get_entities_by_md5(md5: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[int, List[Dict[str, Any]]]]

Implements: https://rest-docs.synapse.org/rest/GET/entity/md5/md5.html

PARAMETER	DESCRIPTION
md5	The MD5 of the entity. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[int, List[Dict[str, Any]]]]	Paginated results of:
Dict[str, Union[int, List[Dict[str, Any]]]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/reflection/model/PaginatedResults.html
Dict[str, Union[int, List[Dict[str, Any]]]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/EntityHeader.html

Source code in synapseclient/api/entity_services.py

async def get_entities_by_md5(
    md5: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[int, List[Dict[str, Any]]]]:
    """
    Implements:
    <https://rest-docs.synapse.org/rest/GET/entity/md5/md5.html>

    Arguments:
        md5: The MD5 of the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        Paginated results of:
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/reflection/model/PaginatedResults.html>
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/EntityHeader.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(
        uri=f"/entity/md5/{md5}",
    )

get_entity_provenance async

get_entity_provenance(entity_id: str, version_number: Optional[int] = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Retrieve provenance information for a Synapse Entity.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. This may include version syn123.0 or syn123. If the version is included in entity_id and version_number is also passed in, then the version in entity_id will be used. TYPE: str
version_number	The version of the Entity to retrieve. Gets the most recent version if omitted. TYPE: Optional[int] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	Activity object as a dictionary or raises exception if no provenance record exists.

Get provenance for an entity

Get the provenance information for entity syn123.

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_entity_provenance

syn = Synapse()
syn.login()

async def main():
    activity = await get_entity_provenance(entity_id="syn123")
    print(f"Activity: {activity}")

asyncio.run(main())

Get provenance for a specific version

Get the provenance information for version 3 of entity syn123.

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_entity_provenance

syn = Synapse()
syn.login()

async def main():
    activity = await get_entity_provenance(entity_id="syn123", version_number=3)
    print(f"Activity: {activity}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def get_entity_provenance(
    entity_id: str,
    version_number: Optional[int] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Retrieve provenance information for a Synapse Entity.

    Arguments:
        entity_id: The ID of the entity. This may include version `syn123.0` or `syn123`.
            If the version is included in `entity_id` and `version_number` is also
            passed in, then the version in `entity_id` will be used.
        version_number: The version of the Entity to retrieve. Gets the most recent version if omitted.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        Activity object as a dictionary or raises exception if no provenance record exists.

    Example: Get provenance for an entity
        Get the provenance information for entity `syn123`.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_entity_provenance

        syn = Synapse()
        syn.login()

        async def main():
            activity = await get_entity_provenance(entity_id="syn123")
            print(f"Activity: {activity}")

        asyncio.run(main())
        ```

    Example: Get provenance for a specific version
        Get the provenance information for version 3 of entity `syn123`.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_entity_provenance

        syn = Synapse()
        syn.login()

        async def main():
            activity = await get_entity_provenance(entity_id="syn123", version_number=3)
            print(f"Activity: {activity}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    syn_id, syn_version = get_synid_and_version(entity_id)
    if not syn_version:
        syn_version = version_number

    if syn_version:
        uri = f"/entity/{syn_id}/version/{syn_version}/generatedBy"
    else:
        uri = f"/entity/{syn_id}/generatedBy"

    return await client.rest_get_async(uri=uri)

set_entity_provenance async

set_entity_provenance(entity_id: str, activity: Dict[str, Any], *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Stores a record of the code and data used to derive a Synapse entity.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
activity	A dictionary representing an Activity object. TYPE: Dict[str, Any]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	An updated Activity object as a dictionary.

Set provenance for an entity

Set the provenance for entity syn123 with an activity.

import asyncio
from synapseclient import Synapse
from synapseclient.api import set_entity_provenance, create_activity

syn = Synapse()
syn.login()

async def main():
    # First create or get an activity
    activity = await create_activity({
        "name": "Analysis Step",
        "description": "Data processing step"
    })

    # Set the provenance
    updated_activity = await set_entity_provenance(
        entity_id="syn123",
        activity=activity
    )
    print(f"Updated activity: {updated_activity}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def set_entity_provenance(
    entity_id: str,
    activity: Dict[str, Any],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Stores a record of the code and data used to derive a Synapse entity.

    Arguments:
        entity_id: The ID of the entity.
        activity: A dictionary representing an Activity object.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        An updated Activity object as a dictionary.

    Example: Set provenance for an entity
        Set the provenance for entity `syn123` with an activity.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import set_entity_provenance, create_activity

        syn = Synapse()
        syn.login()

        async def main():
            # First create or get an activity
            activity = await create_activity({
                "name": "Analysis Step",
                "description": "Data processing step"
            })

            # Set the provenance
            updated_activity = await set_entity_provenance(
                entity_id="syn123",
                activity=activity
            )
            print(f"Updated activity: {updated_activity}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    if "id" in activity:
        saved_activity = await update_activity(activity, synapse_client=synapse_client)
    else:
        saved_activity = await create_activity(activity, synapse_client=synapse_client)

    uri = f"/entity/{entity_id}/generatedBy?generatedBy={saved_activity['id']}"
    return await client.rest_put_async(uri=uri)

delete_entity_provenance async

delete_entity_provenance(entity_id: str, *, synapse_client: Optional[Synapse] = None) -> None

Removes provenance information from an Entity and deletes the associated Activity.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Delete provenance for an entity

Delete the provenance for entity syn123.

import asyncio
from synapseclient import Synapse
from synapseclient.api import delete_entity_provenance

syn = Synapse()
syn.login()

async def main():
    await delete_entity_provenance(entity_id="syn123")

asyncio.run(main())

Returns: None

Source code in synapseclient/api/entity_services.py

async def delete_entity_provenance(
    entity_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> None:
    """
    Removes provenance information from an Entity and deletes the associated Activity.

    Arguments:
        entity_id: The ID of the entity.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Example: Delete provenance for an entity
        Delete the provenance for entity `syn123`.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import delete_entity_provenance

        syn = Synapse()
        syn.login()

        async def main():
            await delete_entity_provenance(entity_id="syn123")

        asyncio.run(main())
        ```

    Returns: None
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    try:
        activity = await get_entity_provenance(entity_id, synapse_client=synapse_client)
    except SynapseHTTPError:
        # If no provenance exists, nothing to delete
        return

    if not activity:
        return

    await client.rest_delete_async(uri=f"/entity/{entity_id}/generatedBy")

    # If the activity is shared by more than one entity you recieve an HTTP 400 error:
    # "If you wish to delete this activity, please first delete all Entities generated by this Activity.""
    await client.rest_delete_async(uri=f"/activity/{activity['id']}")

create_activity async

create_activity(activity: Dict[str, Any], *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Create a new Activity in Synapse.

PARAMETER	DESCRIPTION
activity	A dictionary representing an Activity object. TYPE: Dict[str, Any]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The created Activity object as a dictionary.

Create a new activity

Create a new activity in Synapse.

import asyncio
from synapseclient import Synapse
from synapseclient.api import create_activity

syn = Synapse()
syn.login()

async def main():
    activity = await create_activity({
        "name": "Data Analysis",
        "description": "Processing raw data"
    })
    print(f"Created activity: {activity}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def create_activity(
    activity: Dict[str, Any],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Create a new Activity in Synapse.

    Arguments:
        activity: A dictionary representing an Activity object.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The created Activity object as a dictionary.

    Example: Create a new activity
        Create a new activity in Synapse.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import create_activity

        syn = Synapse()
        syn.login()

        async def main():
            activity = await create_activity({
                "name": "Data Analysis",
                "description": "Processing raw data"
            })
            print(f"Created activity: {activity}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_post_async(uri="/activity", body=json.dumps(activity))

update_activity async

update_activity(activity: Dict[str, Any], *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Modifies an existing Activity.

PARAMETER	DESCRIPTION
activity	The Activity to be updated. Must contain an 'id' field. TYPE: Dict[str, Any]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	An updated Activity object as a dictionary.

RAISES	DESCRIPTION
ValueError	If the activity does not contain an 'id' field.

Update an existing activity

Update an existing activity in Synapse.

import asyncio
from synapseclient import Synapse
from synapseclient.api import update_activity

syn = Synapse()
syn.login()

async def main():
    activity = {
        "id": "12345",
        "name": "Updated Analysis",
        "description": "Updated processing step"
    }
    updated_activity = await update_activity(activity)
    print(f"Updated activity: {updated_activity}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def update_activity(
    activity: Dict[str, Any],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Modifies an existing Activity.

    Arguments:
        activity: The Activity to be updated. Must contain an 'id' field.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        An updated Activity object as a dictionary.

    Raises:
        ValueError: If the activity does not contain an 'id' field.

    Example: Update an existing activity
        Update an existing activity in Synapse.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import update_activity

        syn = Synapse()
        syn.login()

        async def main():
            activity = {
                "id": "12345",
                "name": "Updated Analysis",
                "description": "Updated processing step"
            }
            updated_activity = await update_activity(activity)
            print(f"Updated activity: {updated_activity}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    if "id" not in activity:
        raise ValueError("The activity you want to update must exist on Synapse")

    client = Synapse.get_client(synapse_client=synapse_client)
    uri = f"/activity/{activity['id']}"
    return await client.rest_put_async(uri=uri, body=json.dumps(activity))

get_activity async

get_activity(activity_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Retrieve an Activity by its ID.

PARAMETER	DESCRIPTION
activity_id	The ID of the activity to retrieve. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	Activity object as a dictionary.

Get activity by ID

Retrieve an activity using its ID.

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_activity

syn = Synapse()
syn.login()

async def main():
    activity = await get_activity(activity_id="12345")
    print(f"Activity: {activity}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def get_activity(
    activity_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Retrieve an Activity by its ID.

    Arguments:
        activity_id: The ID of the activity to retrieve.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        Activity object as a dictionary.

    Example: Get activity by ID
        Retrieve an activity using its ID.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_activity

        syn = Synapse()
        syn.login()

        async def main():
            activity = await get_activity(activity_id="12345")
            print(f"Activity: {activity}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(uri=f"/activity/{activity_id}")

get_children async

get_children(parent: Optional[str] = None, include_types: List[str] = None, sort_by: str = 'NAME', sort_direction: str = 'ASC', *, synapse_client: Optional[Synapse] = None) -> AsyncGenerator[Dict[str, Any], None]

Retrieve all entities stored within a parent such as folder or project.

PARAMETER	DESCRIPTION
parent	The ID of a Synapse container (folder or project) or None to retrieve all projects TYPE: Optional[str] DEFAULT: None
include_types	List of entity types to include (e.g., ["folder", "file"]). Available types can be found at: https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/EntityType.html TYPE: List[str] DEFAULT: None
sort_by	How results should be sorted. Can be "NAME" or "CREATED_ON" TYPE: str DEFAULT: 'NAME'
sort_direction	The direction of the result sort. Can be "ASC" or "DESC" TYPE: str DEFAULT: 'ASC'
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

YIELDS	DESCRIPTION
AsyncGenerator[Dict[str, Any], None]	An async generator that yields entity children dictionaries.

Getting children of a folder

Retrieve all children of a folder:

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_children

syn = Synapse()
syn.login()

async def main():
    async for child in get_children(parent="syn123456"):
        print(f"Child: {child['name']} (ID: {child['id']})")

asyncio.run(main())

Getting children with specific types

Retrieve only files and folders:

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_children

syn = Synapse()
syn.login()

async def main():
    async for child in get_children(
        parent="syn123456",
        include_types=["file", "folder"],
        sort_by="NAME",
        sort_direction="ASC"
    ):
        print(f"Child: {child['name']} (Type: {child['type']})")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def get_children(
    parent: Optional[str] = None,
    include_types: List[str] = None,
    sort_by: str = "NAME",
    sort_direction: str = "ASC",
    *,
    synapse_client: Optional["Synapse"] = None,
) -> AsyncGenerator[Dict[str, Any], None]:
    """
    Retrieve all entities stored within a parent such as folder or project.

    Arguments:
        parent: The ID of a Synapse container (folder or project) or None to retrieve all projects
        include_types: List of entity types to include (e.g., ["folder", "file"]).
                      Available types can be found at:
                      https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/EntityType.html
        sort_by: How results should be sorted. Can be "NAME" or "CREATED_ON"
        sort_direction: The direction of the result sort. Can be "ASC" or "DESC"
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor.

    Yields:
        An async generator that yields entity children dictionaries.

    Example: Getting children of a folder
        Retrieve all children of a folder:

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_children

        syn = Synapse()
        syn.login()

        async def main():
            async for child in get_children(parent="syn123456"):
                print(f"Child: {child['name']} (ID: {child['id']})")

        asyncio.run(main())
        ```

    Example: Getting children with specific types
        Retrieve only files and folders:

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_children

        syn = Synapse()
        syn.login()

        async def main():
            async for child in get_children(
                parent="syn123456",
                include_types=["file", "folder"],
                sort_by="NAME",
                sort_direction="ASC"
            ):
                print(f"Child: {child['name']} (Type: {child['type']})")

        asyncio.run(main())
        ```
    """
    if include_types is None:
        include_types = [
            "folder",
            "file",
            "table",
            "link",
            "entityview",
            "dockerrepo",
            "submissionview",
            "dataset",
            "materializedview",
        ]

    request_body = {
        "parentId": parent,
        "includeTypes": include_types,
        "sortBy": sort_by,
        "sortDirection": sort_direction,
    }

    response = rest_post_paginated_async(
        uri="/entity/children",
        body=request_body,
        synapse_client=synapse_client,
    )

    async for child in response:
        yield child

get_child async

get_child(entity_name: str, parent_id: Optional[str] = None, *, synapse_client: Optional[Synapse] = None) -> Optional[str]

Retrieve an entityId for a given parent ID and entity name.

This service can also be used to lookup projectId by setting the parentId to None.

This calls to the REST API found here: https://rest-docs.synapse.org/rest/POST/entity/child.html

PARAMETER	DESCRIPTION
entity_name	The name of the entity to find TYPE: str
parent_id	The parent ID. Set to None when looking up a project by name. TYPE: Optional[str] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Optional[str]	The entity ID if found, None if not found.

RAISES	DESCRIPTION
SynapseHTTPError	If there's an error other than "not found" (404).

Getting a child entity ID

Find a file by name within a folder:

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_child

syn = Synapse()
syn.login()

async def main():
    entity_id = await get_child(
        entity_name="my_file.txt",
        parent_id="syn123456"
    )
    if entity_id:
        print(f"Found entity: {entity_id}")
    else:
        print("Entity not found")

asyncio.run(main())

Getting a project by name

Find a project by name:

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_child

syn = Synapse()
syn.login()

async def main():
    project_id = await get_child(
        entity_name="My Project",
        parent_id=None  # None for projects
    )
    if project_id:
        print(f"Found project: {project_id}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def get_child(
    entity_name: str,
    parent_id: Optional[str] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Optional[str]:
    """
    Retrieve an entityId for a given parent ID and entity name.

    This service can also be used to lookup projectId by setting the parentId to None.

    This calls to the REST API found here: <https://rest-docs.synapse.org/rest/POST/entity/child.html>

    Arguments:
        entity_name: The name of the entity to find
        parent_id: The parent ID. Set to None when looking up a project by name.
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor.

    Returns:
        The entity ID if found, None if not found.

    Raises:
        SynapseHTTPError: If there's an error other than "not found" (404).

    Example: Getting a child entity ID
        Find a file by name within a folder:

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_child

        syn = Synapse()
        syn.login()

        async def main():
            entity_id = await get_child(
                entity_name="my_file.txt",
                parent_id="syn123456"
            )
            if entity_id:
                print(f"Found entity: {entity_id}")
            else:
                print("Entity not found")

        asyncio.run(main())
        ```

    Example: Getting a project by name
        Find a project by name:

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_child

        syn = Synapse()
        syn.login()

        async def main():
            project_id = await get_child(
                entity_name="My Project",
                parent_id=None  # None for projects
            )
            if project_id:
                print(f"Found project: {project_id}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    entity_lookup_request = {
        "parentId": parent_id,
        "entityName": entity_name,
    }

    try:
        response = await client.rest_post_async(
            uri="/entity/child", body=json.dumps(entity_lookup_request)
        )
        return response.get("id")
    except SynapseHTTPError as e:
        if e.response.status_code == 404:
            # Entity not found
            return None
        raise

set_entity_permissions async

set_entity_permissions(entity_id: str, principal_id: Optional[str] = None, access_type: Optional[List[str]] = None, modify_benefactor: bool = False, warn_if_inherits: bool = True, overwrite: bool = True, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]

Set permissions for a user or group on an entity.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
principal_id	Identifier of a user or group. '273948' is for all registered Synapse users and '273949' is for public access. None implies public access. TYPE: Optional[str] DEFAULT: None
access_type	Type of permission to be granted. One or more of CREATE, READ, DOWNLOAD, UPDATE, DELETE, CHANGE_PERMISSIONS. If None or empty list, removes permissions. TYPE: Optional[List[str]] DEFAULT: None
modify_benefactor	Set as True when modifying a benefactor's ACL. TYPE: bool DEFAULT: False
warn_if_inherits	When modify_benefactor is False, and warn_if_inherits is True, a warning log message is produced if the benefactor for the entity you passed into the function is not itself. TYPE: bool DEFAULT: True
overwrite	By default this function overwrites existing permissions for the specified user. Set this flag to False to add new permissions non-destructively. TYPE: bool DEFAULT: True
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]	The updated ACL matching
https	//rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html TYPE: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]

Set permissions for an entity

Grant all registered users download access.

import asyncio
from synapseclient import Synapse
from synapseclient.api import set_entity_permissions

syn = Synapse()
syn.login()

async def main():
    # Grant all registered users download access
    acl = await set_entity_permissions(
        entity_id="syn123",
        principal_id="273948",
        access_type=["READ", "DOWNLOAD"]
    )
    print(f"Updated ACL: {acl}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def set_entity_permissions(
    entity_id: str,
    principal_id: Optional[str] = None,
    access_type: Optional[List[str]] = None,
    modify_benefactor: bool = False,
    warn_if_inherits: bool = True,
    overwrite: bool = True,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]:
    """
    Set permissions for a user or group on an entity.

    Arguments:
        entity_id: The ID of the entity.
        principal_id: Identifier of a user or group. '273948' is for all registered Synapse users
                     and '273949' is for public access. None implies public access.
        access_type: Type of permission to be granted. One or more of CREATE, READ, DOWNLOAD, UPDATE,
                    DELETE, CHANGE_PERMISSIONS. If None or empty list, removes permissions.
        modify_benefactor: Set as True when modifying a benefactor's ACL.
        warn_if_inherits: When modify_benefactor is False, and warn_if_inherits is True,
                         a warning log message is produced if the benefactor for the entity
                         you passed into the function is not itself.
        overwrite: By default this function overwrites existing permissions for the specified user.
                  Set this flag to False to add new permissions non-destructively.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The updated ACL matching
        https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html

    Example: Set permissions for an entity
        Grant all registered users download access.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import set_entity_permissions

        syn = Synapse()
        syn.login()

        async def main():
            # Grant all registered users download access
            acl = await set_entity_permissions(
                entity_id="syn123",
                principal_id="273948",
                access_type=["READ", "DOWNLOAD"]
            )
            print(f"Updated ACL: {acl}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    # Get the benefactor for the entity
    benefactor = await get_entity_benefactor(entity_id, synapse_client=synapse_client)

    if benefactor.id != entity_id:
        if modify_benefactor:
            entity_id = benefactor.id
        elif warn_if_inherits:
            client.logger.warning(
                f"Creating an ACL for entity {entity_id}, which formerly inherited access control from a"
                f' benefactor entity, "{benefactor.name}" ({benefactor.id}).'
            )

    try:
        acl = await get_entity_acl_with_benefactor(
            entity_id=entity_id, synapse_client=synapse_client
        )
    except SynapseHTTPError as e:
        if (
            "The requested ACL does not exist. This entity inherits its permissions from:"
            in str(e)
        ):
            acl = {"resourceAccess": []}
        else:
            raise e

    # Get the principal ID as an integer
    from synapseclient.api.user_services import get_user_by_principal_id_or_name

    principal_id_int = await get_user_by_principal_id_or_name(
        principal_id=principal_id, synapse_client=synapse_client
    )

    # Find existing permissions for this principal
    permissions_to_update = None
    for permissions in acl["resourceAccess"]:
        if (
            "principalId" in permissions
            and permissions["principalId"] == principal_id_int
        ):
            permissions_to_update = permissions
            break

    if access_type is None or access_type == []:
        # Remove permissions
        if permissions_to_update and overwrite:
            acl["resourceAccess"].remove(permissions_to_update)
    else:
        # Add or update permissions
        if not permissions_to_update:
            permissions_to_update = {"accessType": [], "principalId": principal_id_int}
            acl["resourceAccess"].append(permissions_to_update)

        if overwrite:
            permissions_to_update["accessType"] = access_type
        else:
            permissions_to_update["accessType"] = list(
                set(permissions_to_update["accessType"]) | set(access_type)
            )

    benefactor_for_store = await get_entity_benefactor(
        entity_id, synapse_client=synapse_client
    )

    if benefactor_for_store.id == entity_id:
        # Entity is its own benefactor, use PUT
        return await put_entity_acl(
            entity_id=entity_id, acl=acl, synapse_client=synapse_client
        )
    else:
        # Entity inherits ACL, use POST to create new ACL
        return await post_entity_acl(
            entity_id=entity_id, acl=acl, synapse_client=synapse_client
        )

get_entity_acl_list async

get_entity_acl_list(entity_id: str, principal_id: Optional[str] = None, check_benefactor: bool = True, *, synapse_client: Optional[Synapse] = None) -> List[str]

Get a list of permissions for a user or group on an entity.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
principal_id	Identifier of a user or group to check permissions for. If None, returns permissions for the current user. TYPE: Optional[str] DEFAULT: None
check_benefactor	If True (default), check the benefactor for the entity to get the ACL. If False, only check the entity itself. TYPE: bool DEFAULT: True
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	A list of access types that the specified principal has on the entity.

Get ACL list for a user

Get the permissions that a user has on an entity.

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_entity_acl_list

syn = Synapse()
syn.login()

async def main():
    # Get permissions for current user
    permissions = await get_entity_acl_list(entity_id="syn123")
    print(f"Current user permissions: {permissions}")

    # Get permissions for specific user
    permissions = await get_entity_acl_list(
        entity_id="syn123",
        principal_id="12345"
    )
    print(f"User 12345 permissions: {permissions}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def get_entity_acl_list(
    entity_id: str,
    principal_id: Optional[str] = None,
    check_benefactor: bool = True,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> List[str]:
    """
    Get a list of permissions for a user or group on an entity.

    Arguments:
        entity_id: The ID of the entity.
        principal_id: Identifier of a user or group to check permissions for.
                     If None, returns permissions for the current user.
        check_benefactor: If True (default), check the benefactor for the entity
                         to get the ACL. If False, only check the entity itself.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        A list of access types that the specified principal has on the entity.

    Example: Get ACL list for a user
        Get the permissions that a user has on an entity.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_entity_acl_list

        syn = Synapse()
        syn.login()

        async def main():
            # Get permissions for current user
            permissions = await get_entity_acl_list(entity_id="syn123")
            print(f"Current user permissions: {permissions}")

            # Get permissions for specific user
            permissions = await get_entity_acl_list(
                entity_id="syn123",
                principal_id="12345"
            )
            print(f"User 12345 permissions: {permissions}")

        asyncio.run(main())
        ```
    """
    from synapseclient import AUTHENTICATED_USERS, PUBLIC
    from synapseclient.api.team_services import get_teams_for_user
    from synapseclient.api.user_services import (
        get_user_bundle,
        get_user_by_principal_id_or_name,
    )

    # Get the ACL for the entity
    acl = await get_entity_acl_with_benefactor(
        entity_id=entity_id,
        check_benefactor=check_benefactor,
        synapse_client=synapse_client,
    )

    # Get the principal ID as an integer (None defaults to PUBLIC)
    principal_id_int = await get_user_by_principal_id_or_name(
        principal_id=principal_id, synapse_client=synapse_client
    )

    # Get teams that the user belongs to
    team_ids = []
    async for team in get_teams_for_user(
        user_id=str(principal_id_int), synapse_client=synapse_client
    ):
        team_ids.append(int(team["id"]))

    user_profile_bundle = await get_user_bundle(
        user_id=principal_id_int, mask=1, synapse_client=synapse_client
    )

    effective_permission_set = set()

    # Loop over all permissions in the returned ACL and add it to the effective_permission_set
    # if the principalId in the ACL matches
    # 1) the one we are looking for,
    # 2) a team the user is a member of,
    # 3) PUBLIC
    # 4) AUTHENTICATED_USERS (if user_profile_bundle exists for the principal_id)
    for permissions in acl["resourceAccess"]:
        if "principalId" in permissions and (
            permissions["principalId"] == principal_id_int
            or permissions["principalId"] in team_ids
            or permissions["principalId"] == PUBLIC
            or (
                permissions["principalId"] == AUTHENTICATED_USERS
                and user_profile_bundle is not None
            )
        ):
            effective_permission_set = effective_permission_set.union(
                permissions["accessType"]
            )
    return list(effective_permission_set)

update_entity_acl async

update_entity_acl(entity_id: str, acl: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]], *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]

Create or update the Access Control List(ACL) for an entity.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
acl	The ACL to be applied to the entity. Should match the format: {'resourceAccess': [ {'accessType': ['READ', 'DOWNLOAD'], 'principalId': 222222} ]} TYPE: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]	The created or updated ACL matching
https	//rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html TYPE: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]

Update entity ACL

Update the ACL for entity syn123.

import asyncio
from synapseclient import Synapse
from synapseclient.api import update_entity_acl

syn = Synapse()
syn.login()

async def main():
    acl = {
        'resourceAccess': [
            {'accessType': ['READ', 'DOWNLOAD'],
             'principalId': 273948}
        ]
    }
    updated_acl = await update_entity_acl(entity_id="syn123", acl=acl)
    print(f"Updated ACL: {updated_acl}")

asyncio.run(main())

Source code in synapseclient/api/entity_services.py

async def update_entity_acl(
    entity_id: str,
    acl: Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, List[Dict[str, Union[int, List[str]]]]]]:
    """
    Create or update the Access Control List(ACL) for an entity.

    Arguments:
        entity_id: The ID of the entity.
        acl: The ACL to be applied to the entity. Should match the format:
            {'resourceAccess': [
                {'accessType': ['READ', 'DOWNLOAD'],
                 'principalId': 222222}
            ]}
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The created or updated ACL matching
        https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html

    Example: Update entity ACL
        Update the ACL for entity `syn123`.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import update_entity_acl

        syn = Synapse()
        syn.login()

        async def main():
            acl = {
                'resourceAccess': [
                    {'accessType': ['READ', 'DOWNLOAD'],
                     'principalId': 273948}
                ]
            }
            updated_acl = await update_entity_acl(entity_id="syn123", acl=acl)
            print(f"Updated ACL: {updated_acl}")

        asyncio.run(main())
        ```
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    # Get the benefactor to determine whether to use PUT or POST
    benefactor = await get_entity_benefactor(
        entity_id=entity_id, synapse_client=synapse_client
    )

    uri = f"/entity/{entity_id}/acl"
    if benefactor.id == entity_id:
        # Entity is its own benefactor, use PUT to update existing ACL
        return await client.rest_put_async(uri=uri, body=json.dumps(acl))
    else:
        # Entity inherits from a benefactor, use POST to create new ACL
        return await client.rest_post_async(uri=uri, body=json.dumps(acl))

synapseclient.api.file_services

This module is responsible for exposing the services defined at: https://rest-docs.synapse.org/rest/#org.sagebionetworks.repo.web.controller.EntityController

Classes

AddPartResponse dataclass

Result of a part add.

ATTRIBUTE	DESCRIPTION
upload_id	The unique identifier of a multi-part request. TYPE: str
part_number	The part number of the add. TYPE: int
add_part_state	The state of this add. TYPE: str
error_message	If the added failed, this will contain the error message of the cause. Will be None when the add is successful. TYPE: str

Source code in synapseclient/api/file_services.py

@dataclass
class AddPartResponse:
    """Result of a part add.

    Attributes:
        upload_id: The unique identifier of a multi-part request.
        part_number: The part number of the add.
        add_part_state: The state of this add.
        error_message: If the added failed, this will contain the error message of the
            cause. Will be None when the add is successful.
    """

    upload_id: str
    part_number: int
    add_part_state: str
    error_message: str

Functions

post_file_multipart async

post_file_multipart(upload_request_payload: Dict[str, Any], force_restart: bool, endpoint: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, str]

https://rest-docs.synapse.org/rest/POST/file/multipart.html

PARAMETER	DESCRIPTION
upload_request_payload	The request matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/MultipartRequest.html TYPE: Dict[str, Any]
force_restart	Optional parameter. When True, any upload state for the given file will be cleared and a new upload will be started. TYPE: bool
endpoint	Server endpoint to call to. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, str]	The requested multipart upload status matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/MultipartUploadStatus.html

Source code in synapseclient/api/file_services.py

async def post_file_multipart(
    upload_request_payload: Dict[str, Any],
    force_restart: bool,
    endpoint: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, str]:
    """
    <https://rest-docs.synapse.org/rest/POST/file/multipart.html>

    Arguments:
        upload_request_payload: The request matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/MultipartRequest.html>
        force_restart: Optional parameter. When True, any upload state for the given
            file will be cleared and a new upload will be started.
        endpoint: Server endpoint to call to.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The requested multipart upload status matching
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/MultipartUploadStatus.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_post_async(
        f"/file/multipart?forceRestart={str(force_restart).lower()}",
        json.dumps(upload_request_payload),
        endpoint=endpoint,
    )

put_file_multipart_add async

put_file_multipart_add(upload_id: str, part_number: int, md5_hex: str, *, synapse_client: Optional[Synapse] = None) -> AddPartResponse

https://rest-docs.synapse.org/rest/PUT/file/multipart/uploadId/add/partNumber.html

PARAMETER	DESCRIPTION
upload_id	The unique identifier of the file upload. TYPE: str
part_number	The part number to add. Must be a number between 1 and 10,000. TYPE: int
md5_hex	The MD5 of the uploaded part represented as a hexadecimal string. If the provided MD5 does not match the MD5 of the uploaded part, the add will fail. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
AddPartResponse	Object matching
AddPartResponse	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/AddPartResponse.html

Source code in synapseclient/api/file_services.py

async def put_file_multipart_add(
    upload_id: str,
    part_number: int,
    md5_hex: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> AddPartResponse:
    """
    <https://rest-docs.synapse.org/rest/PUT/file/multipart/uploadId/add/partNumber.html>

    Arguments:
        upload_id: The unique identifier of the file upload.
        part_number: The part number to add. Must be a number between 1 and 10,000.
        md5_hex: The MD5 of the uploaded part represented as a hexadecimal string. If
            the provided MD5 does not match the MD5 of the uploaded part, the add
            will fail.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        Object matching
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/AddPartResponse.html>
    """
    try:
        from synapseclient import Synapse

        client = Synapse.get_client(synapse_client=synapse_client)
        response = await client.rest_put_async(
            f"/file/multipart/{upload_id}/add/{part_number}?partMD5Hex={md5_hex}",
            endpoint=client.fileHandleEndpoint,
        )
        return AddPartResponse(
            upload_id=response.get("uploadId", None),
            part_number=response.get("partNumber", None),
            add_part_state=response.get("addPartState", None),
            error_message=response.get("errorMessage", None),
        )
    except Exception:
        client.logger.exception(
            f"Error adding part {part_number} to upload {upload_id} with MD5: {md5_hex}"
        )

put_file_multipart_complete async

put_file_multipart_complete(upload_id: str, endpoint: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, str]

https://rest-docs.synapse.org/rest/PUT/file/multipart/uploadId/complete.html

PARAMETER	DESCRIPTION
upload_id	The unique identifier of the file upload. TYPE: str
endpoint	Server endpoint to call to. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, str]	Object matching
Dict[str, str]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/MultipartUploadStatus.html

Source code in synapseclient/api/file_services.py

async def put_file_multipart_complete(
    upload_id: str,
    endpoint: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, str]:
    """
    <https://rest-docs.synapse.org/rest/PUT/file/multipart/uploadId/complete.html>

    Arguments:
        upload_id: The unique identifier of the file upload.
        endpoint: Server endpoint to call to.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        Object matching
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/MultipartUploadStatus.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_put_async(
        f"/file/multipart/{upload_id}/complete",
        endpoint=endpoint,
    )

post_file_multipart_presigned_urls async

post_file_multipart_presigned_urls(upload_id: str, part_numbers: List[int], *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

https://rest-docs.synapse.org/rest/PUT/file/multipart/uploadId/add/partNumber.html

PARAMETER	DESCRIPTION
upload_id	The unique identifier of the file upload. TYPE: str
part_numbers	The part numbers to get pre-signed URLs for. TYPE: List[int]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	Object matching
Dict[str, Any]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/AddPartResponse.html

Source code in synapseclient/api/file_services.py

async def post_file_multipart_presigned_urls(
    upload_id: str,
    part_numbers: List[int],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    <https://rest-docs.synapse.org/rest/PUT/file/multipart/uploadId/add/partNumber.html>

    Arguments:
        upload_id: The unique identifier of the file upload.
        part_numbers: The part numbers to get pre-signed URLs for.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        Object matching
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/AddPartResponse.html>
    """
    from synapseclient import Synapse

    uri = f"/file/multipart/{upload_id}/presigned/url/batch"
    body = {
        "uploadId": upload_id,
        "partNumbers": part_numbers,
    }

    client = Synapse.get_client(synapse_client=synapse_client)
    client.logger.debug(
        f"Fetching presigned URLs for {part_numbers} with upload_id {upload_id}"
    )
    return await client.rest_post_async(
        uri,
        json.dumps(body),
        endpoint=client.fileHandleEndpoint,
    )

post_external_object_store_filehandle async

post_external_object_store_filehandle(s3_file_key: str, file_path: str, storage_location_id: int, mimetype: str = None, md5: str = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, int]]

Create a new FileHandle representing an external object. https://rest-docs.synapse.org/rest/POST/externalFileHandle.html

PARAMETER	DESCRIPTION
s3_file_key	S3 key of the uploaded object TYPE: str
file_path	The local path of the uploaded file TYPE: str
storage_location_id	The optional storage location descriptor TYPE: int
mimetype	The Mimetype of the file, if known. TYPE: str DEFAULT: None
md5	The file's content MD5, if known. TYPE: str DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, int]]	A FileHandle for objects that are stored externally.
Dict[str, Union[str, int]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/ExternalFileHandleInterface.html

Source code in synapseclient/api/file_services.py

async def post_external_object_store_filehandle(
    s3_file_key: str,
    file_path: str,
    storage_location_id: int,
    mimetype: str = None,
    md5: str = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, int]]:
    """
    Create a new FileHandle representing an external object.
    <https://rest-docs.synapse.org/rest/POST/externalFileHandle.html>

    Arguments:
        s3_file_key:         S3 key of the uploaded object
        file_path:           The local path of the uploaded file
        storage_location_id: The optional storage location descriptor
        mimetype:            The Mimetype of the file, if known.
        md5:                 The file's content MD5, if known.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        A FileHandle for objects that are stored externally.
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/ExternalFileHandleInterface.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    if mimetype is None:
        mimetype, _ = mimetypes.guess_type(file_path, strict=False)
    file_handle = {
        "concreteType": concrete_types.EXTERNAL_OBJECT_STORE_FILE_HANDLE,
        "fileKey": s3_file_key,
        "fileName": os.path.basename(file_path),
        "contentMd5": md5 or utils.md5_for_file(file_path).hexdigest(),
        "contentSize": os.stat(file_path).st_size,
        "storageLocationId": storage_location_id,
        "contentType": mimetype,
    }

    return await client.rest_post_async(
        "/externalFileHandle", json.dumps(file_handle), client.fileHandleEndpoint
    )

post_external_filehandle async

post_external_filehandle(external_url: str, mimetype: str = None, md5: str = None, file_size: int = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, int]]

Create a new FileHandle representing an external object. https://rest-docs.synapse.org/rest/POST/externalFileHandle.html

PARAMETER	DESCRIPTION
externalURL	An external URL
mimetype	The Mimetype of the file, if known. TYPE: str DEFAULT: None
md5	The file's content MD5. TYPE: str DEFAULT: None
file_size	The size of the file in bytes. TYPE: int DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, int]]	A FileHandle for objects that are stored externally.
Dict[str, Union[str, int]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/ExternalFileHandleInterface.html

Source code in synapseclient/api/file_services.py

async def post_external_filehandle(
    external_url: str,
    mimetype: str = None,
    md5: str = None,
    file_size: int = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, int]]:
    """
    Create a new FileHandle representing an external object.
    <https://rest-docs.synapse.org/rest/POST/externalFileHandle.html>

    Arguments:
        externalURL:  An external URL
        mimetype:     The Mimetype of the file, if known.
        md5:          The file's content MD5.
        file_size:    The size of the file in bytes.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        A FileHandle for objects that are stored externally.
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/ExternalFileHandleInterface.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    file_name = external_url.split("/")[-1]
    external_url = utils.as_url(external_url)
    file_handle = {
        "concreteType": concrete_types.EXTERNAL_FILE_HANDLE,
        "fileName": file_name,
        "externalURL": external_url,
        "contentMd5": md5,
        "contentSize": file_size,
    }
    if mimetype is None:
        (mimetype, _) = mimetypes.guess_type(external_url, strict=False)
    if mimetype is not None:
        file_handle["contentType"] = mimetype
    return await client.rest_post_async(
        "/externalFileHandle", json.dumps(file_handle), client.fileHandleEndpoint
    )

post_external_s3_file_handle async

post_external_s3_file_handle(bucket_name: str, s3_file_key: str, file_path: str, parent: str = None, storage_location_id: str = None, mimetype: str = None, md5: str = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, int, bool]]

Create an external S3 file handle for e.g. a file that has been uploaded directly to an external S3 storage location.

https://rest-docs.synapse.org/rest/POST/externalFileHandle/s3.html

PARAMETER	DESCRIPTION
bucket_name	Name of the S3 bucket TYPE: str
s3_file_key	S3 key of the uploaded object TYPE: str
file_path	Local path of the uploaded file TYPE: str
parent	Parent entity to create the file handle in, the file handle will be created in the default storage location of the parent. Mutually exclusive with storage_location_id TYPE: str DEFAULT: None
storage_location_id	Explicit storage location id to create the file handle in, mutually exclusive with parent TYPE: str DEFAULT: None
mimetype	Mimetype of the file, if known TYPE: str DEFAULT: None
md5	MD5 of the file, if known TYPE: str DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, int, bool]]	The created file handle.
Dict[str, Union[str, int, bool]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/S3FileHandle.html

RAISES	DESCRIPTION
ValueError	If neither parent nor storage_location_id is specified, or if both are specified.

Source code in synapseclient/api/file_services.py

async def post_external_s3_file_handle(
    bucket_name: str,
    s3_file_key: str,
    file_path: str,
    parent: str = None,
    storage_location_id: str = None,
    mimetype: str = None,
    md5: str = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, int, bool]]:
    """
    Create an external S3 file handle for e.g. a file that has been uploaded directly to
    an external S3 storage location.

    <https://rest-docs.synapse.org/rest/POST/externalFileHandle/s3.html>

    Arguments:
        bucket_name: Name of the S3 bucket
        s3_file_key: S3 key of the uploaded object
        file_path: Local path of the uploaded file
        parent: Parent entity to create the file handle in, the file handle will be
            created in the default storage location of the parent. Mutually exclusive
            with storage_location_id
        storage_location_id: Explicit storage location id to create the file handle in,
            mutually exclusive with parent
        mimetype: Mimetype of the file, if known
        md5: MD5 of the file, if known
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The created file handle.
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/S3FileHandle.html>

    Raises:
        ValueError: If neither parent nor storage_location_id is specified, or if
            both are specified.
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    if storage_location_id:
        if parent:
            raise ValueError("Pass parent or storage_location_id, not both")
    elif not parent:
        raise ValueError("One of parent or storage_location_id is required")
    else:
        upload_destination = await get_upload_destination(
            entity_id=parent, synapse_client=client
        )
        storage_location_id = upload_destination["storageLocationId"]

    if mimetype is None:
        mimetype, _ = mimetypes.guess_type(file_path, strict=False)

    file_handle = {
        "concreteType": concrete_types.S3_FILE_HANDLE,
        "key": s3_file_key,
        "bucketName": bucket_name,
        "fileName": os.path.basename(file_path),
        "contentMd5": md5 or utils.md5_for_file(file_path).hexdigest(),
        "contentSize": os.stat(file_path).st_size,
        "storageLocationId": storage_location_id,
        "contentType": mimetype,
    }

    return await client.rest_post_async(
        "/externalFileHandle/s3",
        json.dumps(file_handle),
        endpoint=client.fileHandleEndpoint,
    )

get_file_handle async

get_file_handle(file_handle_id: Dict[str, Union[str, int]], *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, int]]

Retrieve a fileHandle from the fileHandle service. Note: You must be the creator of the filehandle to use this method. Otherwise, an 403-Forbidden error will be raised.

https://rest-docs.synapse.org/rest/GET/fileHandle/handleId.html

PARAMETER	DESCRIPTION
file_handle_id	The ID of the file handle to look up. TYPE: Dict[str, Union[str, int]]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, int]]	A file handle retrieved from the file handle service.
Dict[str, Union[str, int]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/FileHandle.html

Source code in synapseclient/api/file_services.py

async def get_file_handle(
    file_handle_id: Dict[str, Union[str, int]],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, int]]:
    """
    Retrieve a fileHandle from the fileHandle service.
    Note: You must be the creator of the filehandle to use this method.
    Otherwise, an 403-Forbidden error will be raised.

    <https://rest-docs.synapse.org/rest/GET/fileHandle/handleId.html>

    Arguments:
        file_handle_id: The ID of the file handle to look up.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        A file handle retrieved from the file handle service.
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/FileHandle.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    return await client.rest_get_async(
        f"/fileHandle/{file_handle_id}", endpoint=client.fileHandleEndpoint
    )

get_file_handle_for_download_async async

get_file_handle_for_download_async(file_handle_id: str, synapse_id: str, entity_type: str = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, str]

Gets the URL and the metadata as filehandle object for a filehandle or fileHandleId

PARAMETER	DESCRIPTION
file_handle_id	ID of fileHandle to download TYPE: str
synapse_id	The ID of the object associated with the file e.g. syn234 TYPE: str
entity_type	Type of object associated with a file e.g. FileEntity, TableEntity https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/FileHandleAssociateType.html TYPE: str DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RAISES	DESCRIPTION
SynapseFileNotFoundError	If the fileHandleId is not found in Synapse.
SynapseError	If the user does not have the permission to access the fileHandleId.

RETURNS	DESCRIPTION
Dict[str, str]	A dictionary with keys: fileHandle, fileHandleId and preSignedURL

Source code in synapseclient/api/file_services.py

async def get_file_handle_for_download_async(
    file_handle_id: str,
    synapse_id: str,
    entity_type: str = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, str]:
    """
    Gets the URL and the metadata as filehandle object for a filehandle or fileHandleId

    Arguments:
        file_handle_id:   ID of fileHandle to download
        synapse_id:       The ID of the object associated with the file e.g. syn234
        entity_type:     Type of object associated with a file e.g. FileEntity,
            TableEntity
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/FileHandleAssociateType.html>
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Raises:
        SynapseFileNotFoundError: If the fileHandleId is not found in Synapse.
        SynapseError: If the user does not have the permission to access the
            fileHandleId.

    Returns:
        A dictionary with keys: fileHandle, fileHandleId and preSignedURL
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    body = {
        "includeFileHandles": True,
        "includePreSignedURLs": True,
        "requestedFiles": [
            {
                "fileHandleId": file_handle_id,
                "associateObjectId": synapse_id,
                "associateObjectType": entity_type or "FileEntity",
            }
        ],
    }
    response = await client.rest_post_async(
        "/fileHandle/batch", body=json.dumps(body), endpoint=client.fileHandleEndpoint
    )

    result = response["requestedFiles"][0]
    failure = result.get("failureCode")
    if failure == "NOT_FOUND":
        raise SynapseFileNotFoundError(
            f"The fileHandleId {file_handle_id} could not be found"
        )
    elif failure == "UNAUTHORIZED":
        raise SynapseAuthorizationError(
            f"You are not authorized to access fileHandleId {file_handle_id} "
            f"associated with the Synapse {entity_type}: {synapse_id}"
        )
    return result

get_file_handle_for_download

get_file_handle_for_download(file_handle_id: str, synapse_id: str, entity_type: str = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, str]

Gets the URL and the metadata as filehandle object for a filehandle or fileHandleId

PARAMETER	DESCRIPTION
file_handle_id	ID of fileHandle to download TYPE: str
synapse_id	The ID of the object associated with the file e.g. syn234 TYPE: str
entity_type	Type of object associated with a file e.g. FileEntity, TableEntity https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/FileHandleAssociateType.html TYPE: str DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RAISES	DESCRIPTION
SynapseFileNotFoundError	If the fileHandleId is not found in Synapse.
SynapseError	If the user does not have the permission to access the fileHandleId.

RETURNS	DESCRIPTION
Dict[str, str]	A dictionary with keys: fileHandle, fileHandleId and preSignedURL

Source code in synapseclient/api/file_services.py

def get_file_handle_for_download(
    file_handle_id: str,
    synapse_id: str,
    entity_type: str = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, str]:
    """
    Gets the URL and the metadata as filehandle object for a filehandle or fileHandleId

    Arguments:
        file_handle_id:   ID of fileHandle to download
        synapse_id:       The ID of the object associated with the file e.g. syn234
        entity_type:     Type of object associated with a file e.g. FileEntity,
            TableEntity
            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/file/FileHandleAssociateType.html>
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Raises:
        SynapseFileNotFoundError: If the fileHandleId is not found in Synapse.
        SynapseError: If the user does not have the permission to access the
            fileHandleId.

    Returns:
        A dictionary with keys: fileHandle, fileHandleId and preSignedURL
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    body = {
        "includeFileHandles": True,
        "includePreSignedURLs": True,
        "requestedFiles": [
            {
                "fileHandleId": file_handle_id,
                "associateObjectId": synapse_id,
                "associateObjectType": entity_type or "FileEntity",
            }
        ],
    }

    response = client.restPOST(
        "/fileHandle/batch", body=json.dumps(body), endpoint=client.fileHandleEndpoint
    )

    result = response["requestedFiles"][0]
    failure = result.get("failureCode")
    if failure == "NOT_FOUND":
        raise SynapseFileNotFoundError(
            f"The fileHandleId {file_handle_id} could not be found"
        )
    elif failure == "UNAUTHORIZED":
        raise SynapseAuthorizationError(
            f"You are not authorized to access fileHandleId {file_handle_id} "
            f"associated with the Synapse {entity_type}: {synapse_id}"
        )
    return result

synapseclient.api.json_schema_services

JSON Schema Services for Synapse

This module provides functions for interacting with JSON schemas in Synapse, including managing organizations, schemas, and entity bindings.

Classes

Functions

bind_json_schema_to_entity async

bind_json_schema_to_entity(synapse_id: str, json_schema_uri: str, *, enable_derived_annotations: bool = False, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

https://rest-docs.synapse.org/rest/PUT/entity/id/schema/binding.html Bind a JSON schema to a Synapse entity.

This creates a binding between an entity and a JSON schema, which enables schema validation for the entity. When bound, the entity's annotations will be validated against the schema requirements.

PARAMETER	DESCRIPTION
synapse_id	The Synapse ID of the entity to bind the schema to TYPE: str
json_schema_uri	The $id URI of the JSON schema to bind (e.g., "my.org-schema.name-1.0.0") TYPE: str
enable_derived_annotations	If True, enables automatic generation of derived annotations from the schema for this entity. Defaults to False. TYPE: bool DEFAULT: False
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	A JsonSchemaObjectBinding object containing the binding details.
Dict[str, Any]	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/JsonSchemaObjectBinding.html

Source code in synapseclient/api/json_schema_services.py

async def bind_json_schema_to_entity(
    synapse_id: str,
    json_schema_uri: str,
    *,
    enable_derived_annotations: bool = False,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    <https://rest-docs.synapse.org/rest/PUT/entity/id/schema/binding.html>
    Bind a JSON schema to a Synapse entity.


    This creates a binding between an entity and a JSON schema, which enables
    schema validation for the entity. When bound, the entity's annotations
    will be validated against the schema requirements.

    Arguments:
        synapse_id: The Synapse ID of the entity to bind the schema to
        json_schema_uri: The $id URI of the JSON schema to bind (e.g., "my.org-schema.name-1.0.0")
        enable_derived_annotations: If True, enables automatic generation of derived annotations
                                   from the schema for this entity. Defaults to False.
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        A JsonSchemaObjectBinding object containing the binding details.
        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/JsonSchemaObjectBinding.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    request_body = {
        "entityId": synapse_id,
        "schema$id": json_schema_uri,
        "enableDerivedAnnotations": enable_derived_annotations,
    }
    return await client.rest_put_async(
        uri=f"/entity/{synapse_id}/schema/binding", body=json.dumps(request_body)
    )

get_json_schema_from_entity async

get_json_schema_from_entity(synapse_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Get the JSON schema binding for a Synapse entity.

https://rest-docs.synapse.org/rest/GET/entity/id/schema/binding.html

Retrieves information about any JSON schema that is currently bound to the specified entity.

PARAMETER	DESCRIPTION
synapse_id	The Synapse ID of the entity to check for schema bindings TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	A JsonSchemaObjectBinding object if a schema is bound, containing:
Dict[str, Any]	entityId: The entity ID
Dict[str, Any]	schema$id: The URI of the bound schema
Dict[str, Any]	enableDerivedAnnotations: Whether derived annotations are enabled
Dict[str, Any]	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/JsonSchemaObjectBinding.html

Source code in synapseclient/api/json_schema_services.py

async def get_json_schema_from_entity(
    synapse_id: str, *, synapse_client: Optional["Synapse"] = None
) -> Dict[str, Any]:
    """
    Get the JSON schema binding for a Synapse entity.

    <https://rest-docs.synapse.org/rest/GET/entity/id/schema/binding.html>


    Retrieves information about any JSON schema that is currently bound to the specified entity.

    Arguments:
        synapse_id: The Synapse ID of the entity to check for schema bindings
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        A JsonSchemaObjectBinding object if a schema is bound, containing:
        - entityId: The entity ID
        - schema$id: The URI of the bound schema
        - enableDerivedAnnotations: Whether derived annotations are enabled

        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/JsonSchemaObjectBinding.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(uri=f"/entity/{synapse_id}/schema/binding")

delete_json_schema_from_entity async

delete_json_schema_from_entity(synapse_id: str, *, synapse_client: Optional[Synapse] = None) -> None

https://rest-docs.synapse.org/rest/DELETE/entity/id/schema/binding.html

Remove the JSON schema binding from a Synapse entity.

This unbinds any JSON schema from the specified entity, removing schema validation requirements and stopping the generation of derived annotations.

PARAMETER	DESCRIPTION
synapse_id	The Synapse ID of the entity to unbind the schema from TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

Source code in synapseclient/api/json_schema_services.py

async def delete_json_schema_from_entity(
    synapse_id: str, *, synapse_client: Optional["Synapse"] = None
) -> None:
    """
    <https://rest-docs.synapse.org/rest/DELETE/entity/id/schema/binding.html>

    Remove the JSON schema binding from a Synapse entity.

    This unbinds any JSON schema from the specified entity, removing schema validation
    requirements and stopping the generation of derived annotations.

    Arguments:
        synapse_id: The Synapse ID of the entity to unbind the schema from
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_delete_async(uri=f"/entity/{synapse_id}/schema/binding")

validate_entity_with_json_schema async

validate_entity_with_json_schema(synapse_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, bool]]

https://rest-docs.synapse.org/rest/GET/entity/id/schema/validation.html

Validate a Synapse entity against its bound JSON schema.

Checks whether the entity's annotations conform to the requirements of its bound JSON schema. The entity must have a schema binding for this operation to work.

PARAMETER	DESCRIPTION
synapse_id	The Synapse ID of the entity to validate TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, bool]]	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/ValidationResults.html
Dict[str, Union[str, bool]]	A ValidationResults object containing:
Dict[str, Union[str, bool]]	objectId: The entity ID that was validated
Dict[str, Union[str, bool]]	objectType: The type of object (typically "entity")
Dict[str, Union[str, bool]]	isValid: Boolean indicating if the entity passes validation
Dict[str, Union[str, bool]]	validatedOn: Timestamp of when validation was performed
Dict[str, Union[str, bool]]	validationErrorMessage: Error details if validation failed
Dict[str, Union[str, bool]]	validationException: Exception details if validation failed

Source code in synapseclient/api/json_schema_services.py

async def validate_entity_with_json_schema(
    synapse_id: str, *, synapse_client: Optional["Synapse"] = None
) -> Dict[str, Union[str, bool]]:
    """
    <https://rest-docs.synapse.org/rest/GET/entity/id/schema/validation.html>

    Validate a Synapse entity against its bound JSON schema.

    Checks whether the entity's annotations conform to the requirements of its bound JSON schema.
    The entity must have a schema binding for this operation to work.

    Arguments:
        synapse_id: The Synapse ID of the entity to validate
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/ValidationResults.html>

        A ValidationResults object containing:
        - objectId: The entity ID that was validated
        - objectType: The type of object (typically "entity")
        - isValid: Boolean indicating if the entity passes validation
        - validatedOn: Timestamp of when validation was performed
        - validationErrorMessage: Error details if validation failed
        - validationException: Exception details if validation failed

    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(uri=f"/entity/{synapse_id}/schema/validation")

get_json_schema_validation_statistics async

get_json_schema_validation_statistics(synapse_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, int]]

Get validation statistics for a container entity (Project or Folder).

Returns summary statistics about JSON schema validation results for all child entities of the specified container that have schema bindings.

https://rest-docs.synapse.org/rest/GET/entity/id/schema/validation/statistics.html

PARAMETER	DESCRIPTION
synapse_id	The Synapse ID of the container entity (Project or Folder) TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, int]]	A ValidationSummaryStatistics object containing:
Dict[str, Union[str, int]]	containerId: The container entity ID
Dict[str, Union[str, int]]	totalNumberOfChildren: Total child entities in the container
Dict[str, Union[str, int]]	numberOfValidChildren: Number of children that pass validation
Dict[str, Union[str, int]]	numberOfInvalidChildren: Number of children that fail validation
Dict[str, Union[str, int]]	numberOfUnknownChildren: Number of children with unknown validation status

Source code in synapseclient/api/json_schema_services.py

async def get_json_schema_validation_statistics(
    synapse_id: str, *, synapse_client: Optional["Synapse"] = None
) -> Dict[str, Union[str, int]]:
    """
    Get validation statistics for a container entity (Project or Folder).

    Returns summary statistics about JSON schema validation results for all child entities
    of the specified container that have schema bindings.

    <https://rest-docs.synapse.org/rest/GET/entity/id/schema/validation/statistics.html>

    Arguments:
        synapse_id: The Synapse ID of the container entity (Project or Folder)
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        A ValidationSummaryStatistics object containing:
        - containerId: The container entity ID
        - totalNumberOfChildren: Total child entities in the container
        - numberOfValidChildren: Number of children that pass validation
        - numberOfInvalidChildren: Number of children that fail validation
        - numberOfUnknownChildren: Number of children with unknown validation status
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(
        uri=f"/entity/{synapse_id}/schema/validation/statistics"
    )

get_invalid_json_schema_validation async

get_invalid_json_schema_validation(synapse_id: str, *, synapse_client: Optional[Synapse] = None) -> AsyncGenerator[Dict[str, Any], None]

https://rest-docs.synapse.org/rest/POST/entity/id/schema/validation/invalid.html

Get all invalid JSON schema validation results for a container entity.

Returns detailed validation results for all child entities of the specified container that fail their JSON schema validation. Results are paginated automatically.

PARAMETER	DESCRIPTION
synapse_id	The Synapse ID of the container entity (Project or Folder) TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

YIELDS	DESCRIPTION
AsyncGenerator[Dict[str, Any], None]	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/ValidationResults.html
AsyncGenerator[Dict[str, Any], None]	ValidationResults objects for each invalid child entity, containing:
AsyncGenerator[Dict[str, Any], None]	objectId: The child entity ID
AsyncGenerator[Dict[str, Any], None]	objectType: The type of object
AsyncGenerator[Dict[str, Any], None]	isValid: Always False for this endpoint
AsyncGenerator[Dict[str, Any], None]	validationErrorMessage: Details about why validation failed
AsyncGenerator[Dict[str, Any], None]	validationException: Exception details

Source code in synapseclient/api/json_schema_services.py

async def get_invalid_json_schema_validation(
    synapse_id: str, *, synapse_client: Optional["Synapse"] = None
) -> AsyncGenerator[Dict[str, Any], None]:
    """
    <https://rest-docs.synapse.org/rest/POST/entity/id/schema/validation/invalid.html>


    Get all invalid JSON schema validation results for a container entity.

    Returns detailed validation results for all child entities of the specified container
    that fail their JSON schema validation. Results are paginated automatically.

    Arguments:
        synapse_id: The Synapse ID of the container entity (Project or Folder)
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Yields:
        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/ValidationResults.html>

        ValidationResults objects for each invalid child entity, containing:
        - objectId: The child entity ID
        - objectType: The type of object
        - isValid: Always False for this endpoint
        - validationErrorMessage: Details about why validation failed
        - validationException: Exception details
    """
    request_body = {"containerId": synapse_id}
    response = rest_post_paginated_async(
        f"/entity/{synapse_id}/schema/validation/invalid",
        body=request_body,
        synapse_client=synapse_client,
    )
    async for item in response:
        yield item

get_invalid_json_schema_validation_sync

get_invalid_json_schema_validation_sync(synapse_id: str, *, synapse_client: Optional[Synapse] = None) -> Generator[Dict[str, Any], None, None]

https://rest-docs.synapse.org/rest/POST/entity/id/schema/validation/invalid.html

Get a single page of invalid JSON schema validation results for a container entity (Project or Folder).

PARAMETER	DESCRIPTION
synapse_id	The Synapse ID of the container entity (Project or Folder) TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

YIELDS	DESCRIPTION
Dict[str, Any]	ValidationResults objects for each invalid child entity.
Dict[str, Any]	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/ValidationResults.html

Source code in synapseclient/api/json_schema_services.py

def get_invalid_json_schema_validation_sync(
    synapse_id: str, *, synapse_client: Optional["Synapse"] = None
) -> Generator[Dict[str, Any], None, None]:
    """
    <https://rest-docs.synapse.org/rest/POST/entity/id/schema/validation/invalid.html>

    Get a single page of invalid JSON schema validation results for a container entity
    (Project or Folder).


    Arguments:
        synapse_id: The Synapse ID of the container entity (Project or Folder)
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Yields:
        ValidationResults objects for each invalid child entity.
        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/ValidationResults.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request_body = {"containerId": synapse_id}
    response = client._POST_paginated(
        f"/entity/{synapse_id}/schema/validation/invalid", request_body
    )
    for item in response:
        yield item

get_json_schema_derived_keys async

get_json_schema_derived_keys(synapse_id: str, *, synapse_client: Optional[Synapse] = None) -> List[str]

https://rest-docs.synapse.org/rest/GET/entity/id/derivedKeys.html

Get the derived annotation keys for a Synapse entity with a bound JSON schema.

When an entity has a JSON schema binding with derived annotations enabled, Synapse can automatically generate annotation keys based on the schema structure. This function retrieves those derived keys.

PARAMETER	DESCRIPTION
synapse_id	The Synapse ID of the entity to get derived keys for TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	A Keys object containing a list of derived annotation key names.
List[str]	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/annotation/v2/Keys.html

Source code in synapseclient/api/json_schema_services.py

async def get_json_schema_derived_keys(
    synapse_id: str, *, synapse_client: Optional["Synapse"] = None
) -> List[str]:
    """
    <https://rest-docs.synapse.org/rest/GET/entity/id/derivedKeys.html>

    Get the derived annotation keys for a Synapse entity with a bound JSON schema.

    When an entity has a JSON schema binding with derived annotations enabled,
    Synapse can automatically generate annotation keys based on the schema structure.
    This function retrieves those derived keys.

    Arguments:
        synapse_id: The Synapse ID of the entity to get derived keys for
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        A Keys object containing a list of derived annotation key names.
        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/annotation/v2/Keys.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    return await client.rest_get_async(uri=f"/entity/{synapse_id}/derivedKeys")

create_organization async

create_organization(organization_name: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Create a new JSON schema organization.

Creates a new organization with a unique name that will serve as a namespace for JSON schemas. The new organization will have an auto-generated AccessControlList (ACL) granting the caller all relevant permissions. Organization names must be at least 6 characters and follow specific naming conventions.

PARAMETER	DESCRIPTION
organization_name	Unique name for the organization. Must be at least 6 characters, cannot start with a number, and should follow dot-separated alphanumeric format (e.g., "my.organization") TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	An Organization object containing:
Dict[str, Any]	id: The numeric identifier of the organization
Dict[str, Any]	name: The organization name
Dict[str, Any]	createdOn: Creation timestamp
Dict[str, Any]	createdBy: ID of the user who created the organization
Dict[str, Any]	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/Organization.html

Source code in synapseclient/api/json_schema_services.py

async def create_organization(
    organization_name: str, *, synapse_client: Optional["Synapse"] = None
) -> Dict[str, Any]:
    """
    Create a new JSON schema organization.

    Creates a new organization with a unique name that will serve as a namespace for JSON schemas.
    The new organization will have an auto-generated AccessControlList (ACL) granting the caller
    all relevant permissions. Organization names must be at least 6 characters and follow specific
    naming conventions.

    Arguments:
        organization_name: Unique name for the organization. Must be at least 6 characters,
                          cannot start with a number, and should follow dot-separated alphanumeric
                          format (e.g., "my.organization")
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        An Organization object containing:
        - id: The numeric identifier of the organization
        - name: The organization name
        - createdOn: Creation timestamp
        - createdBy: ID of the user who created the organization

        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/Organization.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request_body = {"organizationName": organization_name}

    return await client.rest_post_async(
        uri="/schema/organization", body=json.dumps(request_body)
    )

get_organization async

get_organization(organization_name: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Get an organization by name.

Looks up an existing JSON schema organization by its unique name.

PARAMETER	DESCRIPTION
organization_name	The name of the organization to retrieve TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	An Organization object containing:
Dict[str, Any]	id: The numeric identifier of the organization
Dict[str, Any]	name: The organization name
Dict[str, Any]	createdOn: Creation timestamp
Dict[str, Any]	createdBy: ID of the user who created the organization
Dict[str, Any]	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/Organization.html

Source code in synapseclient/api/json_schema_services.py

async def get_organization(
    organization_name: str, *, synapse_client: Optional["Synapse"] = None
) -> Dict[str, Any]:
    """
    Get an organization by name.

    Looks up an existing JSON schema organization by its unique name.

    Arguments:
        organization_name: The name of the organization to retrieve
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        An Organization object containing:
        - id: The numeric identifier of the organization
        - name: The organization name
        - createdOn: Creation timestamp
        - createdBy: ID of the user who created the organization

        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/Organization.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    return await client.rest_get_async(
        uri=f"/schema/organization?name={organization_name}"
    )

list_organizations async

list_organizations(*, synapse_client: Optional[Synapse] = None) -> AsyncGenerator[Dict[str, Any], None]

Generator to list all JSON schema organizations.

Retrieves a list of all organizations that are visible to the caller. This operation does not require authentication and will return all publicly visible organizations.

PARAMETER	DESCRIPTION
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
AsyncGenerator[Dict[str, Any], None]	A generator of Organization objects, each containing:
AsyncGenerator[Dict[str, Any], None]	id: The numeric identifier of the organization
AsyncGenerator[Dict[str, Any], None]	name: The organization name
AsyncGenerator[Dict[str, Any], None]	createdOn: Creation timestamp
AsyncGenerator[Dict[str, Any], None]	createdBy: ID of the user who created the organization
AsyncGenerator[Dict[str, Any], None]	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/Organization.html

Source code in synapseclient/api/json_schema_services.py

async def list_organizations(
    *, synapse_client: Optional["Synapse"] = None
) -> AsyncGenerator[Dict[str, Any], None]:
    """
    Generator to list all JSON schema organizations.

    Retrieves a list of all organizations that are visible to the caller. This operation
    does not require authentication and will return all publicly visible organizations.

    Arguments:
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        A generator of Organization objects, each containing:
        - id: The numeric identifier of the organization
        - name: The organization name
        - createdOn: Creation timestamp
        - createdBy: ID of the user who created the organization

        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/Organization.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request_body = {}

    async for item in rest_post_paginated_async(
        "/schema/organization/list", body=request_body, synapse_client=client
    ):
        yield item

list_organizations_sync

list_organizations_sync(*, synapse_client: Optional[Synapse] = None) -> Generator[Dict[str, Any], None, None]

Generator to list all JSON schema organizations.

Retrieves a list of all organizations that are visible to the caller. This operation does not require authentication and will return all publicly visible organizations.

PARAMETER	DESCRIPTION
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
None	A generator of Organization objects, each containing:
None	id: The numeric identifier of the organization
None	name: The organization name
None	createdOn: Creation timestamp
None	createdBy: ID of the user who created the organization
None	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/Organization.html

Source code in synapseclient/api/json_schema_services.py

def list_organizations_sync(
    *, synapse_client: Optional["Synapse"] = None
) -> Generator[Dict[str, Any], None, None]:
    """
    Generator to list all JSON schema organizations.

    Retrieves a list of all organizations that are visible to the caller. This operation
    does not require authentication and will return all publicly visible organizations.

    Arguments:
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        A generator of Organization objects, each containing:
        - id: The numeric identifier of the organization
        - name: The organization name
        - createdOn: Creation timestamp
        - createdBy: ID of the user who created the organization

        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/Organization.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request_body = {}

    for item in client._POST_paginated("/schema/organization/list", body=request_body):
        yield item

delete_organization async

delete_organization(organization_id: str, *, synapse_client: Optional[Synapse] = None) -> None

Delete a JSON schema organization.

Deletes the specified organization. All schemas defined within the organization's namespace must be deleted first before the organization can be deleted. The caller must have ACCESS_TYPE.DELETE permission on the organization.

PARAMETER	DESCRIPTION
organization_id	The numeric identifier of the organization to delete TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

Source code in synapseclient/api/json_schema_services.py

async def delete_organization(
    organization_id: str, *, synapse_client: Optional["Synapse"] = None
) -> None:
    """
    Delete a JSON schema organization.

    Deletes the specified organization. All schemas defined within the organization's
    namespace must be deleted first before the organization can be deleted. The caller
    must have ACCESS_TYPE.DELETE permission on the organization.

    Arguments:
        organization_id: The numeric identifier of the organization to delete
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    await client.rest_delete_async(uri=f"/schema/organization/{organization_id}")

get_organization_acl async

get_organization_acl(organization_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Get the Access Control List (ACL) for a JSON schema organization.

Retrieves the permissions associated with the specified organization. The caller must have ACCESS_TYPE.READ permission on the organization to view its ACL.

PARAMETER	DESCRIPTION
organization_id	The numeric identifier of the organization TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	An AccessControlList object containing:
Dict[str, Any]	id: The organization ID
Dict[str, Any]	creationDate: The date the ACL was created
Dict[str, Any]	etag: The etag for concurrency control
Dict[str, Any]	resourceAccess: List of ResourceAccess objects with principalId and accessType arrays matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/ResourceAccess.html

Source code in synapseclient/api/json_schema_services.py

async def get_organization_acl(
    organization_id: str, *, synapse_client: Optional["Synapse"] = None
) -> Dict[str, Any]:
    """
    Get the Access Control List (ACL) for a JSON schema organization.

    Retrieves the permissions associated with the specified organization. The caller
    must have ACCESS_TYPE.READ permission on the organization to view its ACL.

    Arguments:
        organization_id: The numeric identifier of the organization
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        An AccessControlList object containing:
        - id: The organization ID
        - creationDate: The date the ACL was created
        - etag: The etag for concurrency control
        - resourceAccess: List of ResourceAccess objects with principalId and accessType arrays matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/ResourceAccess.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    return await client.rest_get_async(
        uri=f"/schema/organization/{organization_id}/acl"
    )

update_organization_acl async

update_organization_acl(organization_id: str, resource_access: Sequence[Mapping[str, Sequence[str]]], etag: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Update the Access Control List (ACL) for a JSON schema organization.

Updates the permissions for the specified organization. The caller must have ACCESS_TYPE.CHANGE_PERMISSIONS permission on the organization. The etag from a previous get_organization_acl() call is required for concurrency control.

PARAMETER	DESCRIPTION
organization_id	The numeric identifier of the organization TYPE: str
resource_access	List of ResourceAccess objects, each containing: - principalId: The user or team ID - accessType: List of permission types (e.g., ["READ", "CREATE", "DELETE"]) TYPE: Sequence[Mapping[str, Sequence[str]]]
etag	The etag from get_organization_acl() for concurrency control TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The updated AccessControlList object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html

Source code in synapseclient/api/json_schema_services.py

async def update_organization_acl(
    organization_id: str,
    resource_access: Sequence[Mapping[str, Sequence[str]]],
    etag: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Any]:
    """
    Update the Access Control List (ACL) for a JSON schema organization.

    Updates the permissions for the specified organization. The caller must have
    ACCESS_TYPE.CHANGE_PERMISSIONS permission on the organization. The etag from
    a previous get_organization_acl() call is required for concurrency control.

    Arguments:
        organization_id: The numeric identifier of the organization
        resource_access: List of ResourceAccess objects, each containing:
                        - principalId: The user or team ID
                        - accessType: List of permission types (e.g., ["READ", "CREATE", "DELETE"])
        etag: The etag from get_organization_acl() for concurrency control
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        The updated AccessControlList object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request_body = {"resourceAccess": resource_access, "etag": etag}

    return await client.rest_put_async(
        uri=f"/schema/organization/{organization_id}/acl", body=json.dumps(request_body)
    )

list_json_schemas async

list_json_schemas(organization_name: str, *, synapse_client: Optional[Synapse] = None) -> AsyncGenerator[Dict[str, Any], None]

List all JSON schemas for an organization.

Retrieves all JSON schemas that belong to the specified organization. This operation does not require authentication and will return all publicly visible schemas.

PARAMETER	DESCRIPTION
organization_name	The name of the organization to list schemas for TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
AsyncGenerator[Dict[str, Any], None]	A generator of JsonSchemaInfo objects, each containing:
AsyncGenerator[Dict[str, Any], None]	organizationId: The Synapse issued numeric identifier for the organization.
AsyncGenerator[Dict[str, Any], None]	organizationName: The name of the organization to which this schema belongs.
AsyncGenerator[Dict[str, Any], None]	schemaId: The Synapse issued numeric identifier for the schema.
AsyncGenerator[Dict[str, Any], None]	schemaName: The name of the this schema.
AsyncGenerator[Dict[str, Any], None]	createdOn: The date this JSON schema was created.
AsyncGenerator[Dict[str, Any], None]	createdBy: The ID of the user that created this JSON schema.

Source code in synapseclient/api/json_schema_services.py

async def list_json_schemas(
    organization_name: str, *, synapse_client: Optional["Synapse"] = None
) -> AsyncGenerator[Dict[str, Any], None]:
    """
    List all JSON schemas for an organization.

    Retrieves all JSON schemas that belong to the specified organization. This operation
    does not require authentication and will return all publicly visible schemas.

    Arguments:
        organization_name: The name of the organization to list schemas for
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        A generator of JsonSchemaInfo objects, each containing:
        - organizationId: The Synapse issued numeric identifier for the organization.
        - organizationName: The name of the organization to which this schema belongs.
        - schemaId: The Synapse issued numeric identifier for the schema.
        - schemaName: The name of the this schema.
        - createdOn: The date this JSON schema was created.
        - createdBy: The ID of the user that created this JSON schema.
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request_body = {"organizationName": organization_name}

    async for item in rest_post_paginated_async(
        "/schema/list", body=request_body, synapse_client=client
    ):
        yield item

list_json_schemas_sync

list_json_schemas_sync(organization_name: str, *, synapse_client: Optional[Synapse] = None) -> Generator[Dict[str, Any], None, None]

List all JSON schemas for an organization.

Retrieves all JSON schemas that belong to the specified organization. This operation does not require authentication and will return all publicly visible schemas.

PARAMETER	DESCRIPTION
organization_name	The name of the organization to list schemas for TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
None	A generator of JsonSchemaInfo objects, each containing:
None	organizationId: The Synapse issued numeric identifier for the organization.
None	organizationName: The name of the organization to which this schema belongs.
None	schemaId: The Synapse issued numeric identifier for the schema.
None	schemaName: The name of the this schema.
None	createdOn: The date this JSON schema was created.
None	createdBy: The ID of the user that created this JSON schema.

Source code in synapseclient/api/json_schema_services.py

def list_json_schemas_sync(
    organization_name: str, *, synapse_client: Optional["Synapse"] = None
) -> Generator[Dict[str, Any], None, None]:
    """
    List all JSON schemas for an organization.

    Retrieves all JSON schemas that belong to the specified organization. This operation
    does not require authentication and will return all publicly visible schemas.

    Arguments:
        organization_name: The name of the organization to list schemas for
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        A generator of JsonSchemaInfo objects, each containing:
        - organizationId: The Synapse issued numeric identifier for the organization.
        - organizationName: The name of the organization to which this schema belongs.
        - schemaId: The Synapse issued numeric identifier for the schema.
        - schemaName: The name of the this schema.
        - createdOn: The date this JSON schema was created.
        - createdBy: The ID of the user that created this JSON schema.
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request_body = {"organizationName": organization_name}

    for item in client._POST_paginated("/schema/list", body=request_body):
        yield item

list_json_schema_versions async

list_json_schema_versions(organization_name: str, json_schema_name: str, *, synapse_client: Optional[Synapse] = None) -> AsyncGenerator[Dict[str, Any], None]

List version information for a JSON schema.

Retrieves version information for all versions of the specified JSON schema within an organization. This shows the history and available versions of a schema.

PARAMETER	DESCRIPTION
organization_name	The name of the organization containing the schema TYPE: str
json_schema_name	The name of the JSON schema to list versions for TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
AsyncGenerator[Dict[str, Any], None]	A generator of JsonSchemaVersionInfo objects, each containing:
AsyncGenerator[Dict[str, Any], None]	organizationId: The Synapse issued numeric identifier for the organization.
AsyncGenerator[Dict[str, Any], None]	organizationName: The name of the organization to which this schema belongs.
AsyncGenerator[Dict[str, Any], None]	schemaName: The name of the this schema.
AsyncGenerator[Dict[str, Any], None]	schemaId: The Synapse issued numeric identifier for the schema.
AsyncGenerator[Dict[str, Any], None]	versionId: The Synapse issued numeric identifier for this version.
AsyncGenerator[Dict[str, Any], None]	$id: The full '$id' of this schema version
AsyncGenerator[Dict[str, Any], None]	semanticVersion: The semantic version label provided when this version was created. Can be null if a semantic version was not provided when this version was created.
AsyncGenerator[Dict[str, Any], None]	createdOn: The date this JSON schema version was created.
AsyncGenerator[Dict[str, Any], None]	createdBy: The ID of the user that created this JSON schema version.
AsyncGenerator[Dict[str, Any], None]	jsonSHA256Hex: The SHA-256 hexadecimal hash of the UTF-8 encoded JSON schema.
AsyncGenerator[Dict[str, Any], None]	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/JsonSchemaVersionInfo.html

Source code in synapseclient/api/json_schema_services.py

async def list_json_schema_versions(
    organization_name: str,
    json_schema_name: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> AsyncGenerator[Dict[str, Any], None]:
    """
    List version information for a JSON schema.

    Retrieves version information for all versions of the specified JSON schema within
    an organization. This shows the history and available versions of a schema.

    Arguments:
        organization_name: The name of the organization containing the schema
        json_schema_name: The name of the JSON schema to list versions for
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        A generator of JsonSchemaVersionInfo objects, each containing:
        - organizationId: The Synapse issued numeric identifier for the organization.
        - organizationName: The name of the organization to which this schema belongs.
        - schemaName: The name of the this schema.
        - schemaId: The Synapse issued numeric identifier for the schema.
        - versionId: The Synapse issued numeric identifier for this version.
        - $id: The full '$id' of this schema version
        - semanticVersion: The semantic version label provided when this version was created. Can be null if a semantic version was not provided when this version was created.
        - createdOn: The date this JSON schema version was created.
        - createdBy: The ID of the user that created this JSON schema version.
        - jsonSHA256Hex: The SHA-256 hexadecimal hash of the UTF-8 encoded JSON schema.

        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/JsonSchemaVersionInfo.html>
    """
    request_body = {
        "organizationName": organization_name,
        "schemaName": json_schema_name,
    }

    async for item in rest_post_paginated_async(
        "/schema/version/list", body=request_body, synapse_client=synapse_client
    ):
        yield item

list_json_schema_versions_sync

list_json_schema_versions_sync(organization_name: str, json_schema_name: str, *, synapse_client: Optional[Synapse] = None) -> Generator[Dict[str, Any], None, None]

List version information for a JSON schema.

Retrieves version information for all versions of the specified JSON schema within an organization. This shows the history and available versions of a schema.

PARAMETER	DESCRIPTION
organization_name	The name of the organization containing the schema TYPE: str
json_schema_name	The name of the JSON schema to list versions for TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
None	A generator of JsonSchemaVersionInfo objects, each containing:
None	organizationId: The Synapse issued numeric identifier for the organization.
None	organizationName: The name of the organization to which this schema belongs.
None	schemaName: The name of the this schema.
None	schemaId: The Synapse issued numeric identifier for the schema.
None	versionId: The Synapse issued numeric identifier for this version.
None	$id: The full '$id' of this schema version
None	semanticVersion: The semantic version label provided when this version was created. Can be null if a semantic version was not provided when this version was created.
None	createdOn: The date this JSON schema version was created.
None	createdBy: The ID of the user that created this JSON schema version.
None	jsonSHA256Hex: The SHA-256 hexadecimal hash of the UTF-8 encoded JSON schema.
None	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/JsonSchemaVersionInfo.html

Source code in synapseclient/api/json_schema_services.py

def list_json_schema_versions_sync(
    organization_name: str,
    json_schema_name: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Generator[Dict[str, Any], None, None]:
    """
    List version information for a JSON schema.

    Retrieves version information for all versions of the specified JSON schema within
    an organization. This shows the history and available versions of a schema.

    Arguments:
        organization_name: The name of the organization containing the schema
        json_schema_name: The name of the JSON schema to list versions for
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        A generator of JsonSchemaVersionInfo objects, each containing:
        - organizationId: The Synapse issued numeric identifier for the organization.
        - organizationName: The name of the organization to which this schema belongs.
        - schemaName: The name of the this schema.
        - schemaId: The Synapse issued numeric identifier for the schema.
        - versionId: The Synapse issued numeric identifier for this version.
        - $id: The full '$id' of this schema version
        - semanticVersion: The semantic version label provided when this version was created. Can be null if a semantic version was not provided when this version was created.
        - createdOn: The date this JSON schema version was created.
        - createdBy: The ID of the user that created this JSON schema version.
        - jsonSHA256Hex: The SHA-256 hexadecimal hash of the UTF-8 encoded JSON schema.

        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/JsonSchemaVersionInfo.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request_body = {
        "organizationName": organization_name,
        "schemaName": json_schema_name,
    }

    for item in client._POST_paginated("/schema/version/list", body=request_body):
        yield item

get_json_schema_body async

get_json_schema_body(json_schema_uri: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Any]

Get a registered JSON schema by its $id URI.

Retrieves the full JSON schema content using its unique $id identifier. This operation does not require authentication for publicly registered schemas.

PARAMETER	DESCRIPTION
json_schema_uri	The relative $id of the JSON schema to get (e.g., "my.org-schema.name-1.0.0") TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	The complete JSON schema object as a dictionary, matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/JsonSchema.html

Source code in synapseclient/api/json_schema_services.py

async def get_json_schema_body(
    json_schema_uri: str, *, synapse_client: Optional["Synapse"] = None
) -> Dict[str, Any]:
    """
    Get a registered JSON schema by its $id URI.

    Retrieves the full JSON schema content using its unique $id identifier. This operation
    does not require authentication for publicly registered schemas.

    Arguments:
        json_schema_uri: The relative $id of the JSON schema to get (e.g., "my.org-schema.name-1.0.0")
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor

    Returns:
        The complete JSON schema object as a dictionary, matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/JsonSchema.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    return await client.rest_get_async(uri=f"/schema/type/registered/{json_schema_uri}")

delete_json_schema async

delete_json_schema(json_schema_uri: str, *, synapse_client: Optional[Synapse] = None) -> None

Delete a JSON schema by its $id URI.

Deletes the specified JSON schema. If the $id excludes a semantic version, all versions of the schema will be deleted. If the $id includes a semantic version, only that specific version will be deleted. The caller must have ACCESS_TYPE.DELETE permission on the schema's organization.

PARAMETER	DESCRIPTION
json_schema_uri	The $id URI of the schema to delete. Examples: - "my.org-schema.name" (deletes all versions) - "my.org-schema.name-1.0.0" (deletes only version 1.0.0) TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

Source code in synapseclient/api/json_schema_services.py

async def delete_json_schema(
    json_schema_uri: str, *, synapse_client: Optional["Synapse"] = None
) -> None:
    """
    Delete a JSON schema by its $id URI.

    Deletes the specified JSON schema. If the $id excludes a semantic version, all versions
    of the schema will be deleted. If the $id includes a semantic version, only that specific
    version will be deleted. The caller must have ACCESS_TYPE.DELETE permission on the
    schema's organization.

    Arguments:
        json_schema_uri: The $id URI of the schema to delete. Examples:
                        - "my.org-schema.name" (deletes all versions)
                        - "my.org-schema.name-1.0.0" (deletes only version 1.0.0)
        synapse_client: If not passed in and caching was not disabled by
                       `Synapse.allow_client_caching(False)` this will use the last created
                       instance from the Synapse class constructor
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    await client.rest_delete_async(uri=f"/schema/type/registered/{json_schema_uri}")

synapseclient.api.table_services

The purpose of this module is to provide any functions that are needed to interact with columns in the Synapse REST API.

Classes

ViewEntityType

Bases: str, Enum

String enum representing the type of view. This is used to determine the default columns that are added to the table. As defined in the Synapse REST API: https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/table/ViewEntityType.html

Source code in synapseclient/api/table_services.py

class ViewEntityType(str, Enum):
    """String enum representing the type of view. This is used to determine
    the default columns that are added to the table.
    As defined in the Synapse REST API:
    <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/table/ViewEntityType.html>
    """

    ENTITY_VIEW = "entityview"
    SUBMISSION_VIEW = "submissionview"
    DATASET = "dataset"
    DATASET_COLLECTION = "datasetcollection"

ViewTypeMask

Bases: int, Enum

Bit mask representing the types to include in the view. As defined in the Synapse REST API: https://rest-docs.synapse.org/rest/GET/column/tableview/defaults.html

Source code in synapseclient/api/table_services.py

class ViewTypeMask(int, Enum):
    """Bit mask representing the types to include in the view.
    As defined in the Synapse REST API:
    <https://rest-docs.synapse.org/rest/GET/column/tableview/defaults.html>
    """

    FILE = 0x01
    PROJECT = 0x02
    TABLE = 0x04
    FOLDER = 0x08
    VIEW = 0x10
    DOCKER = 0x20
    SUBMISSION_VIEW = 0x40
    DATASET = 0x80
    DATASET_COLLECTION = 0x100
    MATERIALIZED_VIEW = 0x200

Functions

create_table_snapshot async

create_table_snapshot(table_id: str, comment: Optional[str] = None, label: Optional[str] = None, activity_id: Optional[str] = None, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, int]]

Creates a table snapshot using the Synapse REST API.

PARAMETER	DESCRIPTION
table_id	Table ID to create a snapshot for. TYPE: str
comment	Optional snapshot comment. TYPE: Optional[str] DEFAULT: None
label	Optional snapshot label. TYPE: Optional[str] DEFAULT: None
activity_id	Optional activity ID or activity instance applied to snapshot version. TYPE: Optional[str] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, int]]	A dictionary containing the snapshot response.

Source code in synapseclient/api/table_services.py

async def create_table_snapshot(
    table_id: str,
    comment: Optional[str] = None,
    label: Optional[str] = None,
    activity_id: Optional[str] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, int]]:
    """
    Creates a table snapshot using the Synapse REST API.

    Arguments:
        table_id: Table ID to create a snapshot for.
        comment: Optional snapshot comment.
        label: Optional snapshot label.
        activity_id: Optional activity ID or activity instance applied to snapshot version.
        synapse_client: If not passed in and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        A dictionary containing the snapshot response.
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    if activity_id and not isinstance(activity_id, str):
        activity_id = str(activity_id)

    snapshot_body = {
        "snapshotComment": comment,
        "snapshotLabel": label,
        "snapshotActivityId": activity_id,
    }

    delete_none_keys(snapshot_body)

    table_id = id_of(table_id)
    uri = f"/entity/{table_id}/table/snapshot"

    snapshot_response = await client.rest_post_async(
        uri, body=json.dumps(snapshot_body)
    )

    return snapshot_response

get_columns async

get_columns(table_id: str, *, synapse_client: Optional[Synapse] = None) -> List[Column]

Get the columns for a Table given the Table's ID.

PARAMETER	DESCRIPTION
table_id	The ID of the Table to get the columns for. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Returns: The annotations set in Synapse.

Source code in synapseclient/api/table_services.py

async def get_columns(
    table_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> List["Column"]:
    """Get the columns for a Table given the Table's ID.

    Arguments:
        table_id: The ID of the Table to get the columns for.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns: The annotations set in Synapse.
    """
    from synapseclient import Synapse
    from synapseclient.models import Column

    result = await Synapse.get_client(synapse_client=synapse_client).rest_get_async(
        f"/entity/{table_id}/column",
    )

    columns = []

    for column in result.get("results", []):
        columns.append(Column().fill_from_dict(synapse_column=column))

    return columns

get_column async

get_column(column_id: str, *, synapse_client: Optional[Synapse] = None) -> Dict[str, Union[str, int]]

Get a single column by ID.

PARAMETER	DESCRIPTION
column_id	The ID of the column to retrieve. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, int]]	The column matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/table/ColumnModel.html

Source code in synapseclient/api/table_services.py

async def get_column(
    column_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict[str, Union[str, int]]:
    """Get a single column by ID.

    Arguments:
        column_id: The ID of the column to retrieve.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The column matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/table/ColumnModel.html>
    """
    from synapseclient import Synapse

    return await Synapse.get_client(synapse_client=synapse_client).rest_get_async(
        f"/column/{column_id}",
    )

list_columns async

list_columns(prefix: Optional[str] = None, limit: int = 100, offset: int = 0, *, synapse_client: Optional[Synapse] = None) -> AsyncGenerator[Column, None]

List columns with optional prefix filtering.

PARAMETER	DESCRIPTION
prefix	Optional prefix to filter columns by name. TYPE: Optional[str] DEFAULT: None
limit	Number of columns to retrieve per request to Synapse (pagination parameter). The function will continue retrieving results until all matching columns are returned. TYPE: int DEFAULT: 100
offset	The index of the first column to return (pagination parameter). TYPE: int DEFAULT: 0
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Yields: Column instances.

Source code in synapseclient/api/table_services.py

async def list_columns(
    prefix: Optional[str] = None,
    limit: int = 100,
    offset: int = 0,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> AsyncGenerator["Column", None]:
    """List columns with optional prefix filtering.

    Arguments:
        prefix: Optional prefix to filter columns by name.
        limit: Number of columns to retrieve per request to Synapse (pagination parameter).
            The function will continue retrieving results until all matching columns are returned.
        offset: The index of the first column to return (pagination parameter).
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Yields: Column instances.
    """
    from synapseclient.api.api_client import rest_get_paginated_async
    from synapseclient.models import Column

    if prefix is None:
        uri = "/column"
    else:
        uri = f"/column?prefix={prefix}"

    async for result in rest_get_paginated_async(
        uri, limit=limit, offset=offset, synapse_client=synapse_client
    ):
        yield Column().fill_from_dict(synapse_column=result)

list_columns_sync

list_columns_sync(prefix: Optional[str] = None, limit: int = 100, offset: int = 0, *, synapse_client: Optional[Synapse] = None) -> Generator[Column, None, None]

List columns with optional prefix filtering (synchronous version).

PARAMETER	DESCRIPTION
prefix	Optional prefix to filter columns by name. TYPE: Optional[str] DEFAULT: None
limit	Number of columns to retrieve per request to Synapse (pagination parameter). The function will continue retrieving results until all matching columns are returned. TYPE: int DEFAULT: 100
offset	The index of the first column to return (pagination parameter). TYPE: int DEFAULT: 0
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Yields: Column instances.

Source code in synapseclient/api/table_services.py

def list_columns_sync(
    prefix: Optional[str] = None,
    limit: int = 100,
    offset: int = 0,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Generator["Column", None, None]:
    """List columns with optional prefix filtering (synchronous version).

    Arguments:
        prefix: Optional prefix to filter columns by name.
        limit: Number of columns to retrieve per request to Synapse (pagination parameter).
            The function will continue retrieving results until all matching columns are returned.
        offset: The index of the first column to return (pagination parameter).
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Yields: Column instances.
    """
    from synapseclient import Synapse
    from synapseclient.models import Column

    client = Synapse.get_client(synapse_client=synapse_client)

    if prefix is None:
        uri = "/column"
    else:
        uri = f"/column?prefix={prefix}"

    for result in client._GET_paginated(uri, limit=limit, offset=offset):
        yield Column().fill_from_dict(synapse_column=result)

post_columns async

post_columns(columns: List[Column], *, synapse_client: Optional[Synapse] = None) -> List[Column]

Creates a batch of synapseclient.models.table.Column's within a single request.

PARAMETER	DESCRIPTION
columns	The columns to post to Synapse. TYPE: List[Column]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor TYPE: Optional[Synapse] DEFAULT: None

Source code in synapseclient/api/table_services.py

async def post_columns(
    columns: List["Column"], *, synapse_client: Optional["Synapse"] = None
) -> List["Column"]:
    """Creates a batch of [synapseclient.models.table.Column][]'s within a single request.

    Arguments:
        columns: The columns to post to Synapse.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor
    """
    from synapseclient import Synapse

    column_values = [column.to_synapse_request() for column in columns]
    request_body = {
        "concreteType": "org.sagebionetworks.repo.model.ListWrapper",
        "list": list(column_values),
    }

    result = await Synapse.get_client(synapse_client=synapse_client).rest_post_async(
        "/column/batch", body=json.dumps(request_body)
    )

    # Fill the results back onto the original columns
    for i, column in enumerate(columns):
        column.fill_from_dict(result["list"][i])

    return columns

get_default_columns async

get_default_columns(view_entity_type: Optional[ViewEntityType] = None, view_type_mask: Optional[int] = None, *, synapse_client: Optional[Synapse] = None) -> List[Column]

Get the default columns for a given view type. This will query the following API: https://rest-docs.synapse.org/rest/GET/column/tableview/defaults.html in order to retrieve this information.

If providing a view_type_mask, you can use the ViewTypeMask enum to get the appropriate value for the entity type you are interested in. ViewTypeMask values are hexadecimal values, so you can use the | operator to combine them.

Example

To get the default columns for a Dataset, you can use:

import asyncio

from synapseclient.api.table_services import ViewEntityType, ViewTypeMask, get_default_columns

async def main():
    view_type_mask = ViewTypeMask.DATASET.value
    columns = await get_default_columns(
        view_entity_type=ViewEntityType.DATASET,
        view_type_mask=view_type_mask,
    )
    print(columns)

asyncio.run(main())

To get the default columns for a File or a Folder, you can use:

import asyncio

from synapseclient.api.table_services import ViewEntityType, ViewTypeMask, get_default_columns

async def main():
    view_type_mask = ViewTypeMask.FILE.value | ViewTypeMask.FOLDER.value
    columns = await get_default_columns(
        view_entity_type=ViewEntityType.DATASET,
        view_type_mask=view_type_mask,
    )
    print(columns)

asyncio.run(main())

PARAMETER	DESCRIPTION
view_type	The type of view to get the default columns for.
view_type_mask	The type of view to get the default columns for. Not required in some cases like a submission view. TYPE: Optional[int] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Returns: The annotations set in Synapse.

Source code in synapseclient/api/table_services.py

async def get_default_columns(
    view_entity_type: Optional[ViewEntityType] = None,
    view_type_mask: Optional[int] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> List["Column"]:
    """Get the default columns for a given view type. This will query the following API:
    <https://rest-docs.synapse.org/rest/GET/column/tableview/defaults.html> in order to
    retrieve this information.

    If providing a view_type_mask, you can use the ViewTypeMask enum to get the
    appropriate value for the entity type you are interested in. ViewTypeMask values are
    hexadecimal values, so you can use the `|` operator to combine them.

    Example:
        To get the default columns for a Dataset, you can use:

        ```python
        import asyncio

        from synapseclient.api.table_services import ViewEntityType, ViewTypeMask, get_default_columns

        async def main():
            view_type_mask = ViewTypeMask.DATASET.value
            columns = await get_default_columns(
                view_entity_type=ViewEntityType.DATASET,
                view_type_mask=view_type_mask,
            )
            print(columns)

        asyncio.run(main())
        ```

        To get the default columns for a File or a Folder, you can use:

        ```python
        import asyncio

        from synapseclient.api.table_services import ViewEntityType, ViewTypeMask, get_default_columns

        async def main():
            view_type_mask = ViewTypeMask.FILE.value | ViewTypeMask.FOLDER.value
            columns = await get_default_columns(
                view_entity_type=ViewEntityType.DATASET,
                view_type_mask=view_type_mask,
            )
            print(columns)

        asyncio.run(main())
        ```

    Arguments:
        view_type: The type of view to get the default columns for.
        view_type_mask: The type of view to get the default columns for. Not required
            in some cases like a submission view.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns: The annotations set in Synapse.
    """

    from synapseclient import Synapse
    from synapseclient.models import Column

    uri = "/column/tableview/defaults"

    if view_entity_type and not view_type_mask:
        uri += f"?viewEntityType={view_entity_type.value}"
    if view_type_mask and not view_entity_type:
        uri += f"?viewTypeMask={view_type_mask}"
    if view_entity_type and view_type_mask:
        uri += f"?viewEntityType={view_entity_type.value}&viewTypeMask={view_type_mask}"

    result = await Synapse.get_client(synapse_client=synapse_client).rest_get_async(uri)

    return [
        Column().fill_from_dict(synapse_column=column)
        for column in result.get("list", [])
    ]

synapseclient.api.team_services

This module is responsible for exposing the services defined at: https://rest-docs.synapse.org/rest/#org.sagebionetworks.repo.web.controller.TeamController

Classes

Functions

post_team_list async

post_team_list(team_ids: List[int], *, synapse_client: Optional[Synapse] = None) -> Optional[List[Dict[str, Union[str, bool]]]]

Retrieve a list of Teams given their IDs. Invalid IDs in the list are ignored: The results list is simply smaller than the list of IDs passed in.

PARAMETER	DESCRIPTION
team_ids	List of team IDs to retrieve TYPE: List[int]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Optional[List[Dict[str, Union[str, bool]]]]	List of dictionaries representing https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Team.html

Source code in synapseclient/api/team_services.py

async def post_team_list(
    team_ids: List[int],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Optional[List[Dict[str, Union[str, bool]]]]:
    """
    Retrieve a list of Teams given their IDs. Invalid IDs in the list are ignored:
    The results list is simply smaller than the list of IDs passed in.

    Arguments:
        team_ids: List of team IDs to retrieve
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        List of dictionaries representing <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Team.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request_body = {"list": team_ids}

    response = await client.rest_post_async(
        uri="/teamList", body=json.dumps(request_body)
    )

    if "list" in response:
        return response["list"] or None

    return None

create_team async

create_team(name: str, description: Optional[str] = None, icon: Optional[str] = None, can_public_join: bool = False, can_request_membership: bool = True, *, synapse_client: Optional[Synapse] = None) -> Dict

Creates a new team.

PARAMETER	DESCRIPTION
name	The name of the team to create. TYPE: str
description	A description of the team. TYPE: Optional[str] DEFAULT: None
icon	The FileHandleID of the icon to be used for the team. TYPE: Optional[str] DEFAULT: None
can_public_join	Whether the team can be joined by anyone. Defaults to False. TYPE: bool DEFAULT: False
can_request_membership	Whether the team can request membership. Defaults to True. TYPE: bool DEFAULT: True
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict	Dictionary representing the created team

Source code in synapseclient/api/team_services.py

async def create_team(
    name: str,
    description: Optional[str] = None,
    icon: Optional[str] = None,
    can_public_join: bool = False,
    can_request_membership: bool = True,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict:
    """
    Creates a new team.

    Arguments:
        name: The name of the team to create.
        description: A description of the team.
        icon: The FileHandleID of the icon to be used for the team.
        can_public_join: Whether the team can be joined by anyone. Defaults to False.
        can_request_membership: Whether the team can request membership. Defaults to True.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        Dictionary representing the created team
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    request_body = {
        "name": name,
        "description": description,
        "icon": icon,
        "canPublicJoin": can_public_join,
        "canRequestMembership": can_request_membership,
    }

    response = await client.rest_post_async(uri="/team", body=json.dumps(request_body))
    return response

delete_team async

delete_team(id: int, *, synapse_client: Optional[Synapse] = None) -> None

Deletes a team.

PARAMETER	DESCRIPTION
id	The ID of the team to delete. TYPE: int
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Source code in synapseclient/api/team_services.py

async def delete_team(
    id: int,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> None:
    """
    Deletes a team.

    Arguments:
        id: The ID of the team to delete.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    await client.rest_delete_async(uri=f"/team/{id}")

get_teams_for_user async

get_teams_for_user(user_id: str, *, synapse_client: Optional[Synapse] = None) -> AsyncGenerator[Dict, None]

Retrieve teams for the matching user ID as an async generator.

This function yields team dictionaries one by one as they are retrieved from the paginated API response, allowing for memory-efficient processing of large result sets.

PARAMETER	DESCRIPTION
user_id	Identifier of a user. TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

YIELDS	DESCRIPTION
AsyncGenerator[Dict, None]	Team dictionaries that the user is a member of. Each dictionary matches the
AsyncGenerator[Dict, None]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Team.html
AsyncGenerator[Dict, None]	structure.

Source code in synapseclient/api/team_services.py

async def get_teams_for_user(
    user_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> AsyncGenerator[Dict, None]:
    """
    Retrieve teams for the matching user ID as an async generator.

    This function yields team dictionaries one by one as they are retrieved from the
    paginated API response, allowing for memory-efficient processing of large result sets.

    Arguments:
        user_id: Identifier of a user.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Yields:
        Team dictionaries that the user is a member of. Each dictionary matches the
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Team.html>
        structure.
    """
    async for result in rest_get_paginated_async(
        uri=f"/user/{user_id}/team", synapse_client=synapse_client
    ):
        yield result

get_team async

get_team(id: Union[int, str], *, synapse_client: Optional[Synapse] = None) -> Dict

Finds a team with a given ID or name.

PARAMETER	DESCRIPTION
id	The ID or name of the team to retrieve. TYPE: Union[int, str]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict	Dictionary representing the team

Source code in synapseclient/api/team_services.py

async def get_team(
    id: Union[int, str],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict:
    """
    Finds a team with a given ID or name.

    Arguments:
        id: The ID or name of the team to retrieve.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        Dictionary representing the team
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    # Retrieves team id
    teamid = id_of(id)
    try:
        int(teamid)
    except (TypeError, ValueError):
        if isinstance(id, str):
            teams = await find_team(id, synapse_client=client)
            for team in teams:
                if team.get("name") == id:
                    teamid = team.get("id")
                    break
            else:
                raise ValueError(f'Can\'t find team "{teamid}"')
        else:
            raise ValueError(f'Can\'t find team "{teamid}"')

    response = await client.rest_get_async(uri=f"/team/{teamid}")
    return response

find_team async

find_team(name: str, *, synapse_client: Optional[Synapse] = None) -> List[Dict]

Retrieve a Teams matching the supplied name fragment

PARAMETER	DESCRIPTION
name	A team name TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[Dict]	List of team dictionaries

Source code in synapseclient/api/team_services.py

async def find_team(
    name: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> List[Dict]:
    """
    Retrieve a Teams matching the supplied name fragment

    Arguments:
        name: A team name
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        List of team dictionaries
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    results = []
    async for result in rest_get_paginated_async(
        uri=f"/teams?fragment={name}", synapse_client=client
    ):
        results.append(result)
    return results

get_team_members async

get_team_members(team: Union[int, str], *, synapse_client: Optional[Synapse] = None) -> List[Dict]

Lists the members of the given team.

PARAMETER	DESCRIPTION
team	A team ID or name. TYPE: Union[int, str]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[Dict]	List of team member dictionaries

Source code in synapseclient/api/team_services.py

async def get_team_members(
    team: Union[int, str],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> List[Dict]:
    """
    Lists the members of the given team.

    Arguments:
        team: A team ID or name.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        List of team member dictionaries
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    team_id = id_of(team)
    results = []
    async for result in rest_get_paginated_async(
        uri=f"/teamMembers/{team_id}", synapse_client=client
    ):
        results.append(result)
    return results

send_membership_invitation async

send_membership_invitation(team_id: int, invitee_id: Optional[str] = None, invitee_email: Optional[str] = None, message: Optional[str] = None, *, synapse_client: Optional[Synapse] = None) -> Dict

Create a membership invitation and send an email notification to the invitee.

PARAMETER	DESCRIPTION
team_id	Synapse team ID TYPE: int
invitee_id	Synapse username or profile id of user TYPE: Optional[str] DEFAULT: None
invitee_email	Email of user TYPE: Optional[str] DEFAULT: None
message	Additional message for the user getting invited to the team. TYPE: Optional[str] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict	MembershipInvitation dictionary

Source code in synapseclient/api/team_services.py

async def send_membership_invitation(
    team_id: int,
    invitee_id: Optional[str] = None,
    invitee_email: Optional[str] = None,
    message: Optional[str] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict:
    """
    Create a membership invitation and send an email notification to the invitee.

    Arguments:
        team_id: Synapse team ID
        invitee_id: Synapse username or profile id of user
        invitee_email: Email of user
        message: Additional message for the user getting invited to the team.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        MembershipInvitation dictionary
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    invite_request = {"teamId": str(team_id), "message": message}
    if invitee_email is not None:
        invite_request["inviteeEmail"] = str(invitee_email)
    if invitee_id is not None:
        invite_request["inviteeId"] = str(invitee_id)

    response = await client.rest_post_async(
        uri="/membershipInvitation", body=json.dumps(invite_request)
    )
    return response

get_team_open_invitations async

get_team_open_invitations(team: Union[int, str], *, synapse_client: Optional[Synapse] = None) -> List[Dict]

Retrieve the open requests submitted to a Team

PARAMETER	DESCRIPTION
team	A team ID or name. TYPE: Union[int, str]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[Dict]	List of MembershipRequest dictionaries

Source code in synapseclient/api/team_services.py

async def get_team_open_invitations(
    team: Union[int, str],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> List[Dict]:
    """
    Retrieve the open requests submitted to a Team

    Arguments:
        team: A team ID or name.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        List of MembershipRequest dictionaries
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    team_id = id_of(team)
    results = []
    async for result in rest_get_paginated_async(
        uri=f"/team/{team_id}/openInvitation", synapse_client=client
    ):
        results.append(result)
    return results

get_membership_status async

get_membership_status(user_id: str, team: Union[int, str], *, synapse_client: Optional[Synapse] = None) -> Dict

Retrieve a user's Team Membership Status bundle.

PARAMETER	DESCRIPTION
user_id	Synapse user ID TYPE: str
team	A team ID or name. TYPE: Union[int, str]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict	Dictionary of TeamMembershipStatus:
Dict	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/TeamMembershipStatus.html

Source code in synapseclient/api/team_services.py

async def get_membership_status(
    user_id: str,
    team: Union[int, str],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict:
    """
    Retrieve a user's Team Membership Status bundle.

    Arguments:
        user_id: Synapse user ID
        team: A team ID or name.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        Dictionary of TeamMembershipStatus:
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/TeamMembershipStatus.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    team_id = id_of(team)
    uri = f"/team/{team_id}/member/{user_id}/membershipStatus"
    response = await client.rest_get_async(uri=uri)
    return response

delete_membership_invitation async

delete_membership_invitation(invitation_id: str, *, synapse_client: Optional[Synapse] = None) -> None

Delete an invitation. Note: The client must be an administrator of the Team referenced by the invitation or the invitee to make this request.

PARAMETER	DESCRIPTION
invitation_id	Open invitation id TYPE: str
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

Source code in synapseclient/api/team_services.py

async def delete_membership_invitation(
    invitation_id: str,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> None:
    """
    Delete an invitation. Note: The client must be an administrator of the
    Team referenced by the invitation or the invitee to make this request.

    Arguments:
        invitation_id: Open invitation id
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    await client.rest_delete_async(uri=f"/membershipInvitation/{invitation_id}")

invite_to_team async

invite_to_team(team: Union[int, str], user: Optional[str] = None, invitee_email: Optional[str] = None, message: Optional[str] = None, force: bool = False, *, synapse_client: Optional[Synapse] = None) -> Optional[Dict]

Invite user to a Synapse team via Synapse username or email (choose one or the other)

PARAMETER	DESCRIPTION
team	A team ID or name. TYPE: Union[int, str]
user	Synapse username or profile id of user TYPE: Optional[str] DEFAULT: None
invitee_email	Email of user TYPE: Optional[str] DEFAULT: None
message	Additional message for the user getting invited to the team. TYPE: Optional[str] DEFAULT: None
force	If an open invitation exists for the invitee, the old invite will be cancelled. TYPE: bool DEFAULT: False
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Optional[Dict]	MembershipInvitation or None if user is already a member

Source code in synapseclient/api/team_services.py

async def invite_to_team(
    team: Union[int, str],
    user: Optional[str] = None,
    invitee_email: Optional[str] = None,
    message: Optional[str] = None,
    force: bool = False,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Optional[Dict]:
    """
    Invite user to a Synapse team via Synapse username or email (choose one or the other)

    Arguments:
        team: A team ID or name.
        user: Synapse username or profile id of user
        invitee_email: Email of user
        message: Additional message for the user getting invited to the team.
        force: If an open invitation exists for the invitee, the old invite will be cancelled.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        MembershipInvitation or None if user is already a member
    """
    from synapseclient.models import UserProfile

    # Input validation
    id_email_specified = invitee_email is not None and user is not None
    id_email_notspecified = invitee_email is None and user is None
    if id_email_specified or id_email_notspecified:
        raise ValueError("Must specify either 'user' or 'inviteeEmail'")

    team_id = id_of(team)
    is_member = False
    open_invitations = await get_team_open_invitations(
        team_id, synapse_client=synapse_client
    )

    if user is not None:
        try:
            user_profile = await UserProfile(username=str(user)).get_async(
                synapse_client=synapse_client
            )
        except SynapseNotFoundError:
            try:
                user_profile = await UserProfile(id=int(user)).get_async(
                    synapse_client=synapse_client
                )
            except (ValueError, TypeError) as ex:
                raise SynapseNotFoundError(f'Can\'t find user "{user}"') from ex
        invitee_id = user_profile.id

        membership_status = await get_membership_status(
            user_id=invitee_id, team=team_id, synapse_client=synapse_client
        )
        is_member = membership_status["isMember"]
        open_invites_to_user = [
            invitation
            for invitation in open_invitations
            if int(invitation.get("inviteeId")) == invitee_id
        ]
    else:
        invitee_id = None
        open_invites_to_user = [
            invitation
            for invitation in open_invitations
            if invitation.get("inviteeEmail") == invitee_email
        ]

    # Only invite if the invitee is not a member and
    # if invitee doesn't have an open invitation unless force=True
    if not is_member and (not open_invites_to_user or force):
        # Delete all old invitations
        for invite in open_invites_to_user:
            await delete_membership_invitation(
                invitation_id=invite["id"], synapse_client=synapse_client
            )
        return await send_membership_invitation(
            team_id,
            invitee_id=invitee_id,
            invitee_email=invitee_email,
            message=message,
            synapse_client=synapse_client,
        )
    else:
        from synapseclient import Synapse

        client = Synapse.get_client(synapse_client=synapse_client)

        if is_member:
            not_sent_reason = f"`{user_profile.username}` is already a member"
        else:
            not_sent_reason = (
                f"`{user_profile.username}` already has an open invitation "
                "Set `force=True` to send new invite."
            )

        client.logger.warning("No invitation sent: {}".format(not_sent_reason))
        return None

synapseclient.api.user_services

This module is responsible for exposing the services defined at: https://rest-docs.synapse.org/rest/#org.sagebionetworks.repo.web.controller.UserGroupController

Classes

Functions

get_user_group_headers_batch async

get_user_group_headers_batch(ids: List[str], *, synapse_client: Optional[Synapse] = None) -> List[Dict[str, Union[str, bool]]]

Batch get UserGroupHeaders. This fetches information about a collection of users or groups, specified by Synapse IDs.

PARAMETER	DESCRIPTION
ids	List of user/group IDs to retrieve TYPE: List[str]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[Dict[str, Union[str, bool]]]	List representing "children" in
List[Dict[str, Union[str, bool]]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/UserGroupHeaderResponsePage.html

Source code in synapseclient/api/user_services.py

async def get_user_group_headers_batch(
    ids: List[str],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> List[Dict[str, Union[str, bool]]]:
    """
    Batch get UserGroupHeaders. This fetches information about a collection of
    users or groups, specified by Synapse IDs.

    Arguments:
        ids: List of user/group IDs to retrieve
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        List representing "children" in
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/UserGroupHeaderResponsePage.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    ids_param = ",".join(ids)
    params = {"ids": ids_param}

    response = await client.rest_get_async(uri="/userGroupHeaders/batch", params=params)

    if "children" in response:
        return response["children"] or []

    return []

get_user_profile_by_id async

get_user_profile_by_id(id: Optional[int] = None, *, synapse_client: Optional[Synapse] = None) -> Dict

Get the details about a Synapse user by ID. Retrieves information on the current user if 'id' is omitted.

PARAMETER	DESCRIPTION
id	The ownerId of a user TYPE: Optional[int] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict	The user profile for the user of interest.

Source code in synapseclient/api/user_services.py

async def get_user_profile_by_id(
    id: Optional[int] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict:
    """
    Get the details about a Synapse user by ID.
    Retrieves information on the current user if 'id' is omitted.

    Arguments:
        id: The ownerId of a user
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The user profile for the user of interest.
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    if id:
        if not isinstance(id, int):
            raise TypeError("id must be an 'ownerId' integer")
    else:
        id = ""

    uri = f"/userProfile/{id}"
    response = await client.rest_get_async(uri=uri)
    return response

get_user_profile_by_username async

get_user_profile_by_username(username: Optional[str] = None, *, synapse_client: Optional[Synapse] = None) -> Dict

Get the details about a Synapse user by username. Retrieves information on the current user if 'username' is omitted or empty string.

PARAMETER	DESCRIPTION
username	The userName of a user TYPE: Optional[str] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict	The user profile for the user of interest.

Source code in synapseclient/api/user_services.py

async def get_user_profile_by_username(
    username: Optional[str] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Dict:
    """
    Get the details about a Synapse user by username.
    Retrieves information on the current user if 'username' is omitted or empty string.

    Arguments:
        username: The userName of a user
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The user profile for the user of interest.
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    is_none = username is None
    is_str = isinstance(username, str)
    if not is_str and not is_none:
        raise TypeError("username must be string or None")

    if is_str:
        principals = await _find_principals(username, synapse_client=synapse_client)
        for principal in principals:
            if principal.get("userName", None).lower() == username.lower():
                id = principal["ownerId"]
                break
        else:
            raise SynapseNotFoundError(f"Can't find user '{username}'")
    else:
        id = ""

    uri = f"/userProfile/{id}"
    response = await client.rest_get_async(uri=uri)
    return response

is_user_certified async

is_user_certified(user: Union[str, int], *, synapse_client: Optional[Synapse] = None) -> bool

Determines whether a Synapse user is a certified user.

PARAMETER	DESCRIPTION
user	Synapse username or Id TYPE: Union[str, int]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
bool	True if the Synapse user is certified

Source code in synapseclient/api/user_services.py

async def is_user_certified(
    user: Union[str, int],
    *,
    synapse_client: Optional["Synapse"] = None,
) -> bool:
    """
    Determines whether a Synapse user is a certified user.

    Arguments:
        user: Synapse username or Id
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        True if the Synapse user is certified
    """

    # Check if userid or username exists - get user profile first
    try:
        # if id is unset or a userID, this will succeed
        user_id = "" if user is None else int(user)
    except (TypeError, ValueError):
        # It's a username, need to look it up
        if isinstance(user, str):
            principals = await _find_principals(user, synapse_client=synapse_client)
            for principal in principals:
                if principal.get("userName", None).lower() == user.lower():
                    user_id = principal["ownerId"]
                    break
            else:  # no break
                raise ValueError(f'Can\'t find user "{user}": ')
        else:
            raise ValueError(f"Invalid user identifier: {user}")

    # Get passing record
    try:
        certification_status = await _get_certified_passing_record(
            user_id, synapse_client=synapse_client
        )
        return certification_status["passed"]
    except SynapseHTTPError as ex:
        if ex.response.status_code == 404:
            # user hasn't taken the quiz
            return False
        raise

get_user_by_principal_id_or_name async

get_user_by_principal_id_or_name(principal_id: Optional[Union[str, int]] = None, *, synapse_client: Optional[Synapse] = None) -> int

Given either a string, int or None finds the corresponding user where None implies PUBLIC.

PARAMETER	DESCRIPTION
principal_id	Identifier of a user or group. '273948' is for all registered Synapse users and '273949' is for public access. None implies public access. TYPE: Optional[Union[str, int]] DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
int	The integer ID of the user.

RAISES	DESCRIPTION
SynapseError	If the user cannot be found or is ambiguous.

Get user ID by principal ID

Get the user ID for a given principal ID.

import asyncio
from synapseclient import Synapse
from synapseclient.api import get_user_by_principal_id_or_name

syn = Synapse()
syn.login()

async def main():
    # Get public user ID
    public_id = await get_user_by_principal_id_or_name(principal_id=None)
    print(f"Public user ID: {public_id}")

    # Get user ID by username
    user_id = await get_user_by_principal_id_or_name(principal_id="username")
    print(f"User ID for 'username': {user_id}")

asyncio.run(main())

Source code in synapseclient/api/user_services.py

async def get_user_by_principal_id_or_name(
    principal_id: Optional[Union[str, int]] = None,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> int:
    """
    Given either a string, int or None finds the corresponding user where None implies PUBLIC.

    Arguments:
        principal_id: Identifier of a user or group. '273948' is for all registered Synapse users
                     and '273949' is for public access. None implies public access.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        The integer ID of the user.

    Raises:
        SynapseError: If the user cannot be found or is ambiguous.

    Example: Get user ID by principal ID
        Get the user ID for a given principal ID.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.api import get_user_by_principal_id_or_name

        syn = Synapse()
        syn.login()

        async def main():
            # Get public user ID
            public_id = await get_user_by_principal_id_or_name(principal_id=None)
            print(f"Public user ID: {public_id}")

            # Get user ID by username
            user_id = await get_user_by_principal_id_or_name(principal_id="username")
            print(f"User ID for 'username': {user_id}")

        asyncio.run(main())
        ```
    """
    from synapseclient import PUBLIC, Synapse

    client = Synapse.get_client(synapse_client=synapse_client)

    if principal_id is None or principal_id == "PUBLIC":
        return PUBLIC
    try:
        return int(principal_id)
    # If principal_id is not a number assume it is a name or email
    except ValueError as ex:
        user_profiles = await client.rest_get_async(
            uri=f"/userGroupHeaders?prefix={principal_id}"
        )
        total_results = len(user_profiles["children"])
        if total_results == 1:
            return int(user_profiles["children"][0]["ownerId"])
        elif total_results > 1:
            for profile in user_profiles["children"]:
                if profile["userName"] == principal_id:
                    return int(profile["ownerId"])

        supplemental_message = (
            "Please be more specific" if total_results > 1 else "No matches"
        )
        raise SynapseError(
            f"Unknown Synapse user ({principal_id}). {supplemental_message}."
        ) from ex

get_user_bundle async

get_user_bundle(user_id: int, mask: int, *, synapse_client: Optional[Synapse] = None) -> Optional[Dict[str, Union[str, dict]]]

Retrieve the user bundle for the given user.

PARAMETER	DESCRIPTION
user_id	Synapse user Id TYPE: int
mask	Bit field indicating which components to include in the bundle. TYPE: int
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Optional[Dict[str, Union[str, dict]]]	Synapse User Bundle or None if user not found
Optional[Dict[str, Union[str, dict]]]	https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/UserBundle.html

Source code in synapseclient/api/user_services.py

async def get_user_bundle(
    user_id: int,
    mask: int,
    *,
    synapse_client: Optional["Synapse"] = None,
) -> Optional[Dict[str, Union[str, dict]]]:
    """
    Retrieve the user bundle for the given user.

    Arguments:
        user_id: Synapse user Id
        mask: Bit field indicating which components to include in the bundle.
        synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

    Returns:
        Synapse User Bundle or None if user not found
        <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/UserBundle.html>
    """
    from synapseclient import Synapse

    client = Synapse.get_client(synapse_client=synapse_client)
    try:
        return await client.rest_get_async(uri=f"/user/{user_id}/bundle?mask={mask}")
    except SynapseHTTPError as ex:
        if ex.response.status_code == 404:
            return None
        raise