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

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.
    """
    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_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_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

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

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) -> Union[Dict[str, Any], str]

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

Bind a JSON schema to an entity

PARAMETER	DESCRIPTION
synapse_id	Synapse Entity or Synapse Id TYPE: str
json_schema_uri	JSON schema URI TYPE: str
enable_derived_annotations	If True, derived annotations will be enabled for this entity 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: 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,
) -> Union[Dict[str, Any], str]:
    """
    <https://rest-docs.synapse.org/rest/PUT/entity/id/schema/binding.html>

    Bind a JSON schema to an entity

    Arguments:
        synapse_id:      Synapse Entity or Synapse Id
        json_schema_uri: JSON schema URI
        enable_derived_annotations:  If True, derived annotations will be enabled for this 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:
        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) -> Union[Dict[str, Any], str]

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

Get bound schema from entity

PARAMETER	DESCRIPTION
synapse_id	Synapse 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

RETURNS	DESCRIPTION
Union[Dict[str, Any], str]	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
) -> Union[Dict[str, Any], str]:
    """
    <https://rest-docs.synapse.org/rest/GET/entity/id/schema/binding.html>

    Get bound schema from entity

    Arguments:
        synapse_id:      Synapse 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:
        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

Delete bound schema from entity

PARAMETER	DESCRIPTION
synapse_id	Synapse 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/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>

    Delete bound schema from entity

    Arguments:
        synapse_id:      Synapse 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)
    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) -> Union[Dict[str, Union[str, bool]], str]

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

Get validation results of an entity against bound JSON schema

PARAMETER	DESCRIPTION
synapse_id	Synapse 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

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

Source code in synapseclient/api/json_schema_services.py

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

    Get validation results of an entity against bound JSON schema

    Arguments:
        synapse_id:      Synapse 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:
        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)
    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]]

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

Get the summary statistic of json schema validation results for a container entity Arguments: synapse_id: Synapse 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	DESCRIPTION
Dict[str, Union[str, int]]	Object matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/ValidationSummaryStatistics.html

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]]:
    """

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

    Get the summary statistic of json schema validation results for
        a container entity
    Arguments:
        synapse_id:      Synapse 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:
        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/ValidationSummaryStatistics.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/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 a single page of invalid JSON schema validation results for a container Entity (Project or Folder).

PARAMETER	DESCRIPTION
synapse_id	Synapse 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

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

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 a single page of invalid JSON schema validation results for a container Entity
    (Project or Folder).

    Arguments:
        synapse_id:      Synapse 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:
        Object matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/ValidationResults.html>
    """

    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	Synapse 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

Returns: 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:      Synapse 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:
        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

Retrieve derived JSON schema keys for a given Synapse entity.

PARAMETER	DESCRIPTION
synapse_id	The Synapse ID of the entity for which to retrieve derived keys. TYPE: str

RETURNS	DESCRIPTION
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>

    Retrieve derived JSON schema keys for a given Synapse entity.

    Arguments:
        synapse_id (str): The Synapse ID of the entity for which to retrieve derived keys.

    Returns:
        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")

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_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: 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