Async Object-Orientated Models

Contained within this file are experimental interfaces for working with the Synapse Python Client. Unless otherwise noted these interfaces are subject to change at any time. Use at your own risk.

These APIs also introduce AsyncIO to the client.

Sample Scripts:

See this page for sample scripts. The sample scripts are from a synchronous context, replace any of the method calls with the async counter-party and they will be functionally equivalent.

API reference

synapseclient.models.Project dataclass

Bases: ProjectSynchronousProtocol, AccessControllable, StorableContainer

A Project is a top-level container for organizing data in Synapse.

ATTRIBUTE	DESCRIPTION
id	The unique immutable ID for this project. A new ID will be generated for new Projects. Once issued, this ID is guaranteed to never change or be re-issued TYPE: Optional[str]
name	The name of this project. Must be 256 characters or less. Names may only contain: letters, numbers, spaces, underscores, hyphens, periods, plus signs, apostrophes, and parentheses TYPE: Optional[str]
description	The description of this entity. Must be 1000 characters or less. TYPE: Optional[str]
etag	Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle concurrent updates. Since the E-Tag changes every time an entity is updated it is used to detect when a client's current representation of an entity is out-of-date. TYPE: Optional[str]
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]
alias	The project alias for use in friendly project urls. TYPE: Optional[str]
files	Any files that are at the root directory of the project. TYPE: Optional[List[File]]
folders	Any folders that are at the root directory of the project. TYPE: Optional[List[Folder]]
annotations	Additional metadata associated with the folder. The key is the name of your desired annotations. The value is an object containing a list of values (use empty list to represent no values for key) and the value type associated with all values in the list. To remove all annotations set this to an empty dict {}. TYPE: Optional[Dict[str, Union[List[str], List[bool], List[float], List[int], List[date], List[datetime]]]]

Creating a project

This example shows how to create a project

from synapseclient.models import Project, File
import synapseclient

synapseclient.login()

my_annotations = {
    "my_single_key_string": "a",
    "my_key_string": ["b", "a", "c"],
}
project = Project(
    name="My unique project name",
    annotations=my_annotations,
    description="This is a project with random data.",
)

project = project.store()

print(project)

Storing several files to a project

This example shows how to store several files to a project

file_1 = File(
    path=path_to_file_1,
    name=name_of_file_1,
)
file_2 = File(
    path=path_to_file_2,
    name=name_of_file_2,
)
project.files = [file_1, file_2]
project = project.store()

Source code in synapseclient/models/project.py

@dataclass()
@async_to_sync
class Project(ProjectSynchronousProtocol, AccessControllable, StorableContainer):
    """A Project is a top-level container for organizing data in Synapse.

    Attributes:
        id: The unique immutable ID for this project. A new ID will be generated for new
            Projects. Once issued, this ID is guaranteed to never change or be re-issued
        name: The name of this project. Must be 256 characters or less. Names may only
            contain: letters, numbers, spaces, underscores, hyphens, periods, plus
            signs, apostrophes, and parentheses
        description: The description of this entity. Must be 1000 characters or less.
        etag: Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
            concurrent updates. Since the E-Tag changes every time an entity is updated
            it is used to detect when a client's current representation of an entity
            is out-of-date.
        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.
        alias: The project alias for use in friendly project urls.
        files: Any files that are at the root directory of the project.
        folders: Any folders that are at the root directory of the project.
        annotations: Additional metadata associated with the folder. The key is the name
            of your desired annotations. The value is an object containing a list of
            values (use empty list to represent no values for key) and the value type
            associated with all values in the list.  To remove all annotations set this
            to an empty dict `{}`.

    Example: Creating a project
        This example shows how to create a project

            from synapseclient.models import Project, File
            import synapseclient

            synapseclient.login()

            my_annotations = {
                "my_single_key_string": "a",
                "my_key_string": ["b", "a", "c"],
            }
            project = Project(
                name="My unique project name",
                annotations=my_annotations,
                description="This is a project with random data.",
            )

            project = project.store()

            print(project)

    Example: Storing several files to a project
        This example shows how to store several files to a project

            file_1 = File(
                path=path_to_file_1,
                name=name_of_file_1,
            )
            file_2 = File(
                path=path_to_file_2,
                name=name_of_file_2,
            )
            project.files = [file_1, file_2]
            project = project.store()

    """

    id: Optional[str] = None
    """The unique immutable ID for this project. A new ID will be generated for new
    Projects. Once issued, this ID is guaranteed to never change or be re-issued"""

    name: Optional[str] = None
    """The name of this project. Must be 256 characters or less. Names may only contain:
    letters, numbers, spaces, underscores, hyphens, periods, plus signs, apostrophes,
    and parentheses"""

    description: Optional[str] = None
    """The description of this entity. Must be 1000 characters or less."""

    etag: Optional[str] = None
    """Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
    concurrent updates. Since the E-Tag changes every time an entity is updated it
    is used to detect when a client's current representation of an entity is out-of-date."""

    created_on: Optional[str] = field(default=None, compare=False)
    """(Read Only) The date this entity was created."""

    modified_on: Optional[str] = field(default=None, compare=False)
    """(Read Only) The date this entity was last modified."""

    created_by: Optional[str] = field(default=None, compare=False)
    """(Read Only) The ID of the user that created this entity."""

    modified_by: Optional[str] = field(default=None, compare=False)
    """(Read Only) The ID of the user that last modified this entity."""

    alias: Optional[str] = None
    """The project alias for use in friendly project urls."""

    files: Optional[List["File"]] = field(default_factory=list, compare=False)
    """Any files that are at the root directory of the project."""

    folders: Optional[List["Folder"]] = field(default_factory=list, compare=False)
    """Any folders that are at the root directory of the project."""

    annotations: Optional[
        Dict[
            str,
            Union[
                List[str],
                List[bool],
                List[float],
                List[int],
                List[date],
                List[datetime],
            ],
        ]
    ] = field(default=None, compare=False)
    """Additional metadata associated with the folder. The key is the name of your
    desired annotations. The value is an object containing a list of values
    (use empty list to represent no values for key) and the value type associated with
    all values in the list. To remove all annotations set this to an empty dict `{}`."""

    create_or_update: bool = field(default=True, repr=False)
    """
    (Store only)

    Indicates whether the method should automatically perform an update if the resource
    conflicts with an existing Synapse object. When True this means that any changes
    to the resource will be non-destructive.

    This boolean is ignored if you've already stored or retrieved the resource from
    Synapse for this instance at least once. Any changes to the resource will be
    destructive in this case. For example if you want to delete the content for a field
    you will need to call `.get()` and then modify the field.
    """

    parent_id: Optional[str] = None
    """The parent ID of the project. In practice projects do not have a parent, but this
    is required for the inner workings of Synapse."""

    _last_persistent_instance: Optional["Project"] = field(
        default=None, repr=False, compare=False
    )
    """The last persistent instance of this object. This is used to determine if the
    object has been changed and needs to be updated in Synapse."""

    @property
    def has_changed(self) -> bool:
        """Determines if the object has been changed and needs to be updated in Synapse."""
        return (
            not self._last_persistent_instance or self._last_persistent_instance != self
        )

    def _set_last_persistent_instance(self) -> None:
        """Stash the last time this object interacted with Synapse. This is used to
        determine if the object has been changed and needs to be updated in Synapse."""
        del self._last_persistent_instance
        self._last_persistent_instance = replace(self)
        self._last_persistent_instance.annotations = (
            deepcopy(self.annotations) if self.annotations else None
        )

    def fill_from_dict(
        self,
        synapse_project: Union[Synapse_Project, Dict],
        set_annotations: bool = True,
    ) -> "Project":
        """
        Converts a response from the REST API into this dataclass.

        Arguments:
            synapse_project: The response from the REST API.

        Returns:
            The Project object.
        """
        self.id = synapse_project.get("id", None)
        self.name = synapse_project.get("name", None)
        self.description = synapse_project.get("description", None)
        self.etag = synapse_project.get("etag", None)
        self.created_on = synapse_project.get("createdOn", None)
        self.modified_on = synapse_project.get("modifiedOn", None)
        self.created_by = synapse_project.get("createdBy", None)
        self.modified_by = synapse_project.get("modifiedBy", None)
        self.alias = synapse_project.get("alias", None)
        self.parent_id = synapse_project.get("parentId", None)
        if set_annotations:
            self.annotations = Annotations.from_dict(
                synapse_project.get("annotations", None)
            )
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Project_Store: ID: {self.id}, Name: {self.name}"
    )
    async def store_async(
        self,
        failure_strategy: FailureStrategy = FailureStrategy.LOG_EXCEPTION,
        synapse_client: Optional[Synapse] = None,
    ) -> "Project":
        """
        Store project, files, and folders to synapse. If you have any files or folders
        attached to this project they will be stored as well. You may attach files
        and folders to this project by setting the `files` and `folders` attributes.

        By default the store operation will non-destructively update the project if
        you have not already retrieved the project from Synapse. If you have already
        retrieved the project from Synapse then the store operation will be destructive
        and will overwrite the project with the current state of this object. See the
        `create_or_update` attribute for more information.

        Arguments:
            failure_strategy: Determines how to handle failures when storing attached
                Files and Folders under this Project and an exception occurs.
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            The project object.

        Example: Using this method to update the description
            Store the project to Synapse using ID

                project = await Project(id="syn123", description="new").store_async()

            Store the project to Synapse using Name

                project = await Project(name="my_project", description="new").store_async()

        Raises:
            ValueError: If the project name is not set.
        """
        if not self.name and not self.id:
            raise ValueError("Project ID or Name is required")

        if (
            self.create_or_update
            and not self._last_persistent_instance
            and (
                existing_project_id := await get_id(
                    entity=self, synapse_client=synapse_client, failure_strategy=None
                )
            )
            and (
                existing_project := await Project(id=existing_project_id).get_async(
                    synapse_client=synapse_client
                )
            )
        ):
            merge_dataclass_entities(source=existing_project, destination=self)

        if self.has_changed:
            loop = asyncio.get_event_loop()
            synapse_project = Synapse_Project(
                id=self.id,
                etag=self.etag,
                name=self.name,
                description=self.description,
                alias=self.alias,
                parentId=self.parent_id,
            )
            delete_none_keys(synapse_project)
            current_context = context.get_current()
            entity = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(synapse_client=synapse_client).store(
                        obj=synapse_project,
                        set_annotations=False,
                        createOrUpdate=False,
                    ),
                    current_context,
                ),
            )
            self.fill_from_dict(synapse_project=entity, set_annotations=False)

        await store_entity_components(
            root_resource=self,
            failure_strategy=failure_strategy,
            synapse_client=synapse_client,
        )

        self._set_last_persistent_instance()
        Synapse.get_client(synapse_client=synapse_client).logger.debug(
            f"Saved Project {self.name}, id: {self.id}"
        )

        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Project_Get: ID: {self.id}, Name: {self.name}"
    )
    async def get_async(
        self,
        synapse_client: Optional[Synapse] = None,
    ) -> "Project":
        """Get the project metadata from Synapse.

        Arguments:
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            The project object.

        Example: Using this method
            Retrieve the project from Synapse using ID

                project = await Project(id="syn123").get_async()

            Retrieve the project from Synapse using Name

                project = await Project(name="my_project").get_async()

        Raises:
            ValueError: If the project ID or Name is not set.
            SynapseNotFoundError: If the project is not found in Synapse.
        """
        entity_id = await get_id(entity=self, synapse_client=synapse_client)

        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        entity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).get(
                    entity=entity_id,
                ),
                current_context,
            ),
        )

        self.fill_from_dict(synapse_project=entity, set_annotations=True)

        self._set_last_persistent_instance()
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Project_Delete: {self.id}, Name: {self.name}"
    )
    async def delete_async(self, synapse_client: Optional[Synapse] = None) -> None:
        """Delete the project from Synapse.

        Arguments:
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            None

        Example: Using this method
            Delete the project from Synapse using ID

                await Project(id="syn123").delete_async()

            Delete the project from Synapse using Name

                await Project(name="my_project").delete_async()

        Raises:
            ValueError: If the project ID or Name is not set.
            SynapseNotFoundError: If the project is not found in Synapse.
        """
        entity_id = await get_id(entity=self, synapse_client=synapse_client)

        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).delete(
                    obj=entity_id,
                ),
                current_context,
            ),
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Project_Copy: {self.id}"
    )
    async def copy_async(
        self,
        destination_id: str,
        copy_annotations: bool = True,
        copy_wiki: bool = True,
        exclude_types: Optional[List[str]] = None,
        file_update_existing: bool = False,
        file_copy_activity: Union[str, None] = "traceback",
        synapse_client: Optional[Synapse] = None,
    ) -> "Project":
        """
        You must have already created the Project you will be copying to. It will have
        it's own Synapse ID and unique name that you will use as the destination_id.


        Copy the project to another Synapse project. This will recursively copy all
        Tables, Links, Files, and Folders within the project.

        Arguments:
            destination_id: Synapse ID of a project to copy to.
            copy_annotations: True to copy the annotations.
            copy_wiki: True to copy the wiki pages.
            exclude_types: A list of entity types ['file', 'table', 'link'] which
                determines which entity types to not copy. Defaults to an empty list.
            file_update_existing: When the destination has a file that has the same
                name, users can choose to update that file.
            file_copy_activity: Has three options to set the activity of the copied file:

                    - traceback: Creates a copy of the source files Activity.
                    - existing: Link to the source file's original Activity (if it exists)
                    - None: No activity is set
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            The copied project object.

        Example: Using this function
            Assuming you have a project with the ID "syn123" and you want to copy it to a
            project with the ID "syn456":

                new_instance = await Project(id="syn123").copy_async(destination_id="syn456")

            Copy the project but do not persist annotations:

                new_instance = await Project(id="syn123").copy_async(destination_id="syn456", copy_annotations=False)

        Raises:
            ValueError: If the project does not have an ID and destination_id to copy.
        """
        if not self.id or not destination_id:
            raise ValueError("The project must have an ID and destination_id to copy.")

        loop = asyncio.get_event_loop()

        current_context = context.get_current()
        syn = Synapse.get_client(synapse_client=synapse_client)
        source_and_destination = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: copy(
                    syn=syn,
                    entity=self.id,
                    destinationId=destination_id,
                    excludeTypes=exclude_types or [],
                    skipCopyAnnotations=not copy_annotations,
                    skipCopyWikiPage=not copy_wiki,
                    updateExisting=file_update_existing,
                    setProvenance=file_copy_activity,
                ),
                current_context,
            ),
        )

        new_project_id = source_and_destination.get(self.id, None)
        if not new_project_id:
            raise SynapseError("Failed to copy project.")
        project_copy = await (
            await Project(id=new_project_id).get_async()
        ).sync_from_synapse_async(
            download_file=False,
            synapse_client=synapse_client,
        )
        Synapse.get_client(synapse_client=synapse_client).logger.debug(
            f"Copied from project {self.id} to {destination_id}"
        )
        return project_copy

Functions

get_async(synapse_client=None) async

Get the project metadata from Synapse.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Project	The project object.

Using this method

Retrieve the project from Synapse using ID

project = await Project(id="syn123").get_async()

Retrieve the project from Synapse using Name

project = await Project(name="my_project").get_async()

RAISES	DESCRIPTION
ValueError	If the project ID or Name is not set.
SynapseNotFoundError	If the project is not found in Synapse.

Source code in synapseclient/models/project.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Project_Get: ID: {self.id}, Name: {self.name}"
)
async def get_async(
    self,
    synapse_client: Optional[Synapse] = None,
) -> "Project":
    """Get the project metadata from Synapse.

    Arguments:
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        The project object.

    Example: Using this method
        Retrieve the project from Synapse using ID

            project = await Project(id="syn123").get_async()

        Retrieve the project from Synapse using Name

            project = await Project(name="my_project").get_async()

    Raises:
        ValueError: If the project ID or Name is not set.
        SynapseNotFoundError: If the project is not found in Synapse.
    """
    entity_id = await get_id(entity=self, synapse_client=synapse_client)

    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    entity = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).get(
                entity=entity_id,
            ),
            current_context,
        ),
    )

    self.fill_from_dict(synapse_project=entity, set_annotations=True)

    self._set_last_persistent_instance()
    return self

store_async(failure_strategy=FailureStrategy.LOG_EXCEPTION, synapse_client=None) async

Store project, files, and folders to synapse. If you have any files or folders attached to this project they will be stored as well. You may attach files and folders to this project by setting the files and folders attributes.

By default the store operation will non-destructively update the project if you have not already retrieved the project from Synapse. If you have already retrieved the project from Synapse then the store operation will be destructive and will overwrite the project with the current state of this object. See the create_or_update attribute for more information.

PARAMETER	DESCRIPTION
failure_strategy	Determines how to handle failures when storing attached Files and Folders under this Project and an exception occurs. TYPE: FailureStrategy DEFAULT: LOG_EXCEPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Project	The project object.

Using this method to update the description

Store the project to Synapse using ID

project = await Project(id="syn123", description="new").store_async()

Store the project to Synapse using Name

project = await Project(name="my_project", description="new").store_async()

RAISES	DESCRIPTION
ValueError	If the project name is not set.

Source code in synapseclient/models/project.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Project_Store: ID: {self.id}, Name: {self.name}"
)
async def store_async(
    self,
    failure_strategy: FailureStrategy = FailureStrategy.LOG_EXCEPTION,
    synapse_client: Optional[Synapse] = None,
) -> "Project":
    """
    Store project, files, and folders to synapse. If you have any files or folders
    attached to this project they will be stored as well. You may attach files
    and folders to this project by setting the `files` and `folders` attributes.

    By default the store operation will non-destructively update the project if
    you have not already retrieved the project from Synapse. If you have already
    retrieved the project from Synapse then the store operation will be destructive
    and will overwrite the project with the current state of this object. See the
    `create_or_update` attribute for more information.

    Arguments:
        failure_strategy: Determines how to handle failures when storing attached
            Files and Folders under this Project and an exception occurs.
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        The project object.

    Example: Using this method to update the description
        Store the project to Synapse using ID

            project = await Project(id="syn123", description="new").store_async()

        Store the project to Synapse using Name

            project = await Project(name="my_project", description="new").store_async()

    Raises:
        ValueError: If the project name is not set.
    """
    if not self.name and not self.id:
        raise ValueError("Project ID or Name is required")

    if (
        self.create_or_update
        and not self._last_persistent_instance
        and (
            existing_project_id := await get_id(
                entity=self, synapse_client=synapse_client, failure_strategy=None
            )
        )
        and (
            existing_project := await Project(id=existing_project_id).get_async(
                synapse_client=synapse_client
            )
        )
    ):
        merge_dataclass_entities(source=existing_project, destination=self)

    if self.has_changed:
        loop = asyncio.get_event_loop()
        synapse_project = Synapse_Project(
            id=self.id,
            etag=self.etag,
            name=self.name,
            description=self.description,
            alias=self.alias,
            parentId=self.parent_id,
        )
        delete_none_keys(synapse_project)
        current_context = context.get_current()
        entity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).store(
                    obj=synapse_project,
                    set_annotations=False,
                    createOrUpdate=False,
                ),
                current_context,
            ),
        )
        self.fill_from_dict(synapse_project=entity, set_annotations=False)

    await store_entity_components(
        root_resource=self,
        failure_strategy=failure_strategy,
        synapse_client=synapse_client,
    )

    self._set_last_persistent_instance()
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Saved Project {self.name}, id: {self.id}"
    )

    return self

delete_async(synapse_client=None) async

Delete the project from Synapse.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
None	None

Using this method

Delete the project from Synapse using ID

await Project(id="syn123").delete_async()

Delete the project from Synapse using Name

await Project(name="my_project").delete_async()

RAISES	DESCRIPTION
ValueError	If the project ID or Name is not set.
SynapseNotFoundError	If the project is not found in Synapse.

Source code in synapseclient/models/project.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Project_Delete: {self.id}, Name: {self.name}"
)
async def delete_async(self, synapse_client: Optional[Synapse] = None) -> None:
    """Delete the project from Synapse.

    Arguments:
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        None

    Example: Using this method
        Delete the project from Synapse using ID

            await Project(id="syn123").delete_async()

        Delete the project from Synapse using Name

            await Project(name="my_project").delete_async()

    Raises:
        ValueError: If the project ID or Name is not set.
        SynapseNotFoundError: If the project is not found in Synapse.
    """
    entity_id = await get_id(entity=self, synapse_client=synapse_client)

    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).delete(
                obj=entity_id,
            ),
            current_context,
        ),
    )

sync_from_synapse_async(path=None, recursive=True, download_file=True, if_collision=COLLISION_OVERWRITE_LOCAL, failure_strategy=FailureStrategy.LOG_EXCEPTION, synapse_client=None) async

Sync this container and all possible sub-folders from Synapse. By default this will download the files that are found and it will populate the files and folders attributes with the found files and folders. If you only want to retrieve the full tree of metadata about your container specify download_file as False.

This works similar to synapseutils.syncFromSynapse, however, this does not currently support the writing of data to a manifest TSV file. This will be a future enhancement.

Only Files and Folders are supported at this time to be synced from synapse.

PARAMETER	DESCRIPTION
path	An optional path where the file hierarchy will be reproduced. If not specified the files will by default be placed in the synapseCache. TYPE: Optional[str] DEFAULT: None
recursive	Whether or not to recursively get the entire hierarchy of the folder and sub-folders. TYPE: bool DEFAULT: True
download_file	Whether to download the files found or not. TYPE: bool DEFAULT: True
if_collision	Determines how to handle file collisions. May be overwrite.local keep.local keep.both TYPE: str DEFAULT: COLLISION_OVERWRITE_LOCAL
failure_strategy	Determines how to handle failures when retrieving children under this Folder and an exception occurs. TYPE: FailureStrategy DEFAULT: LOG_EXCEPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Self	The object that was called on. This will be the same object that was called on to start the sync.

Using this function

Suppose I want to walk the immediate children of a folder without downloading the files:

from synapseclient import Synapse
from synapseclient.models import Folder

syn = Synapse()
syn.login()

my_folder = Folder(id="syn12345")
await my_folder.sync_from_synapse_async(download_file=False, recursive=False)

for folder in my_folder.folders:
    print(folder.name)

for file in my_folder.files:
    print(file.name)

Suppose I want to download the immediate children of a folder:

from synapseclient import Synapse
from synapseclient.models import Folder

syn = Synapse()
syn.login()

my_folder = Folder(id="syn12345")
await my_folder.sync_from_synapse_async(path="/path/to/folder", recursive=False)

for folder in my_folder.folders:
    print(folder.name)

for file in my_folder.files:
    print(file.name)

Suppose I want to download the immediate all children of a Project and all sub-folders and files:

from synapseclient import Synapse
from synapseclient.models import Project

syn = Synapse()
syn.login()

my_project = Project(id="syn12345")
await my_project.sync_from_synapse_async(path="/path/to/folder")

RAISES	DESCRIPTION
ValueError	If the folder does not have an id set.

A sequence diagram for this method is as follows:

sequenceDiagram
    autonumber
    participant project_or_folder
    activate project_or_folder
    project_or_folder->>sync_from_synapse: Recursive search and download files
    activate sync_from_synapse
        opt Current instance not retrieved from Synapse
            sync_from_synapse->>project_or_folder: Call `.get()` method
            project_or_folder-->>sync_from_synapse: .
        end

        loop For each return of the generator
            sync_from_synapse->>client: call `.getChildren()` method
            client-->>sync_from_synapse: .
            note over sync_from_synapse: Append to a running list
        end

        loop For each child
            note over sync_from_synapse: Create all `pending_tasks` at current depth

            alt Child is File
                note over sync_from_synapse: Append `file.get()` method
            else Child is Folder
                note over sync_from_synapse: Append `folder.get()` method
                alt Recursive is True
                    note over sync_from_synapse: Append `folder.sync_from_synapse()` method
                end
            end
        end

        loop For each task in pending_tasks
            par `file.get()`
                sync_from_synapse->>File: Retrieve File metadata and Optionally download
                File->>client: `.get()`
                client-->>File: .
                File-->>sync_from_synapse: .
            and `folder.get()`
                sync_from_synapse->>Folder: Retrieve Folder metadataa
                Folder->>client: `.get()`
                client-->>Folder: .
                Folder-->>sync_from_synapse: .
            and `folder.sync_from_synapse_async()`
                note over sync_from_synapse: This is a recursive call to `sync_from_synapse`
                sync_from_synapse->>sync_from_synapse: Recursive call to `.sync_from_synapse_async()`
            end
        end

    deactivate sync_from_synapse
    deactivate project_or_folder

Source code in synapseclient/models/mixins/storable_container.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"{self.__class__.__name__}_sync_from_synapse: {self.id}"
)
async def sync_from_synapse_async(
    self: Self,
    path: Optional[str] = None,
    recursive: bool = True,
    download_file: bool = True,
    if_collision: str = COLLISION_OVERWRITE_LOCAL,
    failure_strategy: FailureStrategy = FailureStrategy.LOG_EXCEPTION,
    synapse_client: Optional[Synapse] = None,
) -> Self:
    """
    Sync this container and all possible sub-folders from Synapse. By default this
    will download the files that are found and it will populate the
    `files` and `folders` attributes with the found files and folders. If you only
    want to retrieve the full tree of metadata about your container specify
    `download_file` as False.

    This works similar to [synapseutils.syncFromSynapse][], however, this does not
    currently support the writing of data to a manifest TSV file. This will be a
    future enhancement.

    Only Files and Folders are supported at this time to be synced from synapse.

    Arguments:
        path: An optional path where the file hierarchy will be reproduced. If not
            specified the files will by default be placed in the synapseCache.
        recursive: Whether or not to recursively get the entire hierarchy of the
            folder and sub-folders.
        download_file: Whether to download the files found or not.
        if_collision: Determines how to handle file collisions. May be

            - `overwrite.local`
            - `keep.local`
            - `keep.both`
        failure_strategy: Determines how to handle failures when retrieving children
            under this Folder and an exception occurs.
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        The object that was called on. This will be the same object that was called on
            to start the sync.

    Example: Using this function
        Suppose I want to walk the immediate children of a folder without downloading the files:

            from synapseclient import Synapse
            from synapseclient.models import Folder

            syn = Synapse()
            syn.login()

            my_folder = Folder(id="syn12345")
            await my_folder.sync_from_synapse_async(download_file=False, recursive=False)

            for folder in my_folder.folders:
                print(folder.name)

            for file in my_folder.files:
                print(file.name)

        Suppose I want to download the immediate children of a folder:

            from synapseclient import Synapse
            from synapseclient.models import Folder

            syn = Synapse()
            syn.login()

            my_folder = Folder(id="syn12345")
            await my_folder.sync_from_synapse_async(path="/path/to/folder", recursive=False)

            for folder in my_folder.folders:
                print(folder.name)

            for file in my_folder.files:
                print(file.name)


        Suppose I want to download the immediate all children of a Project and all sub-folders and files:

            from synapseclient import Synapse
            from synapseclient.models import Project

            syn = Synapse()
            syn.login()

            my_project = Project(id="syn12345")
            await my_project.sync_from_synapse_async(path="/path/to/folder")


    Raises:
        ValueError: If the folder does not have an id set.


    A sequence diagram for this method is as follows:

    ```mermaid
    sequenceDiagram
        autonumber
        participant project_or_folder
        activate project_or_folder
        project_or_folder->>sync_from_synapse: Recursive search and download files
        activate sync_from_synapse
            opt Current instance not retrieved from Synapse
                sync_from_synapse->>project_or_folder: Call `.get()` method
                project_or_folder-->>sync_from_synapse: .
            end

            loop For each return of the generator
                sync_from_synapse->>client: call `.getChildren()` method
                client-->>sync_from_synapse: .
                note over sync_from_synapse: Append to a running list
            end

            loop For each child
                note over sync_from_synapse: Create all `pending_tasks` at current depth

                alt Child is File
                    note over sync_from_synapse: Append `file.get()` method
                else Child is Folder
                    note over sync_from_synapse: Append `folder.get()` method
                    alt Recursive is True
                        note over sync_from_synapse: Append `folder.sync_from_synapse()` method
                    end
                end
            end

            loop For each task in pending_tasks
                par `file.get()`
                    sync_from_synapse->>File: Retrieve File metadata and Optionally download
                    File->>client: `.get()`
                    client-->>File: .
                    File-->>sync_from_synapse: .
                and `folder.get()`
                    sync_from_synapse->>Folder: Retrieve Folder metadataa
                    Folder->>client: `.get()`
                    client-->>Folder: .
                    Folder-->>sync_from_synapse: .
                and `folder.sync_from_synapse_async()`
                    note over sync_from_synapse: This is a recursive call to `sync_from_synapse`
                    sync_from_synapse->>sync_from_synapse: Recursive call to `.sync_from_synapse_async()`
                end
            end

        deactivate sync_from_synapse
        deactivate project_or_folder
    ```

    """
    if not self._last_persistent_instance:
        await self.get_async(synapse_client=synapse_client)
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Syncing {self.__class__.__name__} ({self.id}) from Synapse."
    )
    path = os.path.expanduser(path) if path else None

    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    children = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: self._retrieve_children(
                synapse_client=synapse_client,
            ),
            current_context,
        ),
    )

    pending_tasks = []
    self.folders = []
    self.files = []

    for child in children:
        pending_tasks.extend(
            self._create_task_for_child(
                child=child,
                recursive=recursive,
                path=path,
                download_file=download_file,
                if_collision=if_collision,
                failure_strategy=failure_strategy,
                synapse_client=synapse_client,
            )
        )

    for task in asyncio.as_completed(pending_tasks):
        result = await task
        self._resolve_sync_from_synapse_result(
            result=result,
            failure_strategy=failure_strategy,
            synapse_client=synapse_client,
        )
    return self

get_permissions_async(synapse_client=None) async

Get the permissions that the caller has on an Entity.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Permissions	A Permissions object

Using this function:

Getting permissions for a Synapse Entity

permissions = await File(id="syn123").get_permissions_async()

Getting access types list from the Permissions object

permissions.access_types

Source code in synapseclient/models/mixins/access_control.py

async def get_permissions_async(
    self,
    synapse_client: Optional[Synapse] = None,
) -> "Permissions":
    """
    Get the [permissions][synapseclient.core.models.permission.Permissions]
    that the caller has on an Entity.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        A Permissions object


    Example: Using this function:
        Getting permissions for a Synapse Entity

            permissions = await File(id="syn123").get_permissions_async()

        Getting access types list from the Permissions object

            permissions.access_types
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).get_permissions(entity=self.id),
            current_context,
        ),
    )

get_acl_async(principal_id=None, synapse_client=None) async

Get the ACL that a user or group has on an Entity.

PARAMETER	DESCRIPTION
principal_id	Identifier of a user or group (defaults to PUBLIC users) TYPE: int DEFAULT: None
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	An array containing some combination of ['READ', 'UPDATE', 'CREATE', 'DELETE', 'DOWNLOAD', 'MODERATE', 'CHANGE_PERMISSIONS', 'CHANGE_SETTINGS'] or an empty array

Source code in synapseclient/models/mixins/access_control.py

async def get_acl_async(
    self, principal_id: int = None, synapse_client: Optional[Synapse] = None
) -> List[str]:
    """
    Get the [ACL][synapseclient.core.models.permission.Permissions.access_types]
    that a user or group has on an Entity.

    Arguments:
        principal_id: Identifier of a user or group (defaults to PUBLIC users)
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        An array containing some combination of
            ['READ', 'UPDATE', 'CREATE', 'DELETE', 'DOWNLOAD', 'MODERATE',
            'CHANGE_PERMISSIONS', 'CHANGE_SETTINGS']
            or an empty array
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).get_acl(
                entity=self.id, principal_id=principal_id
            ),
            current_context,
        ),
    )

set_permissions_async(principal_id=None, access_type=None, modify_benefactor=False, warn_if_inherits=True, overwrite=True, synapse_client=None) async

Sets permission that a user or group has on an Entity. An Entity may have its own ACL or inherit its ACL from a benefactor.

PARAMETER	DESCRIPTION
principal_id	Identifier of a user or group. 273948 is for all registered Synapse users and 273949 is for public access. None implies public access. TYPE: int DEFAULT: None
access_type	Type of permission to be granted. One or more of CREATE, READ, DOWNLOAD, UPDATE, DELETE, CHANGE_PERMISSIONS. Defaults to ['READ', 'DOWNLOAD'] TYPE: List[str] DEFAULT: None
modify_benefactor	Set as True when modifying a benefactor's ACL TYPE: bool DEFAULT: False
warn_if_inherits	Set as False, when creating a new ACL. Trying to modify the ACL of an Entity that inherits its ACL will result in a warning TYPE: bool DEFAULT: True
overwrite	By default this function overwrites existing permissions for the specified user. Set this flag to False to add new permissions non-destructively. TYPE: bool DEFAULT: True
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, list]]	An Access Control List object

Setting permissions

Grant all registered users download access

await File(id="syn123").set_permissions_async(principal_id=273948, access_type=['READ','DOWNLOAD'])

Grant the public view access

await File(id="syn123").set_permissions_async(principal_id=273949, access_type=['READ'])

Source code in synapseclient/models/mixins/access_control.py

async def set_permissions_async(
    self,
    principal_id: int = None,
    access_type: List[str] = None,
    modify_benefactor: bool = False,
    warn_if_inherits: bool = True,
    overwrite: bool = True,
    synapse_client: Optional[Synapse] = None,
) -> Dict[str, Union[str, list]]:
    """
    Sets permission that a user or group has on an Entity.
    An Entity may have its own ACL or inherit its ACL from a benefactor.

    Arguments:
        principal_id: Identifier of a user or group. `273948` is for all
            registered Synapse users and `273949` is for public access.
            None implies public access.
        access_type: Type of permission to be granted. One or more of CREATE,
            READ, DOWNLOAD, UPDATE, DELETE, CHANGE_PERMISSIONS.

            **Defaults to ['READ', 'DOWNLOAD']**
        modify_benefactor: Set as True when modifying a benefactor's ACL
        warn_if_inherits: Set as False, when creating a new ACL. Trying to modify
            the ACL of an Entity that inherits its ACL will result in a warning
        overwrite: By default this function overwrites existing permissions for
            the specified user. Set this flag to False to add new permissions
            non-destructively.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        An Access Control List object

    Example: Setting permissions
        Grant all registered users download access

            await File(id="syn123").set_permissions_async(principal_id=273948, access_type=['READ','DOWNLOAD'])

        Grant the public view access

            await File(id="syn123").set_permissions_async(principal_id=273949, access_type=['READ'])
    """
    if access_type is None:
        access_type = ["READ", "DOWNLOAD"]
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).setPermissions(
                entity=self.id,
                principalId=principal_id,
                accessType=access_type,
                modify_benefactor=modify_benefactor,
                warn_if_inherits=warn_if_inherits,
                overwrite=overwrite,
            ),
            current_context,
        ),
    )

---

synapseclient.models.Folder dataclass

Bases: FolderSynchronousProtocol, AccessControllable, StorableContainer

Folder is a hierarchical container for organizing data in Synapse.

ATTRIBUTE	DESCRIPTION
id	The unique immutable ID for this folder. A new ID will be generated for new Folders. Once issued, this ID is guaranteed to never change or be re-issued. TYPE: Optional[str]
name	The name of this folder. Must be 256 characters or less. Names may only contain: letters, numbers, spaces, underscores, hyphens, periods, plus signs, apostrophes, and parentheses. TYPE: Optional[str]
parent_id	The ID of the Project or Folder that is the parent of this Folder. TYPE: Optional[str]
description	The description of this entity. Must be 1000 characters or less. TYPE: Optional[str]
etag	(Read Only) Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle concurrent updates. Since the E-Tag changes every time an entity is updated it is used to detect when a client's current representation of an entity is out-of-date. TYPE: Optional[str]
created_on	(Read Only) The date this entity was created. TYPE: Optional[str]
modified_on	(Read Only) The date this entity was last modified. TYPE: Optional[str]
created_by	(Read Only) The ID of the user that created this entity. TYPE: Optional[str]
modified_by	(Read Only) The ID of the user that last modified this entity. TYPE: Optional[str]
files	Files that exist within this folder. TYPE: Optional[List[File]]
folders	Folders that exist within this folder. TYPE: Optional[List[Folder]]
annotations	Additional metadata associated with the folder. The key is the name of your desired annotations. The value is an object containing a list of values (use empty list to represent no values for key) and the value type associated with all values in the list. To remove all annotations set this to an empty dict {}. TYPE: Optional[Dict[str, Union[List[str], List[bool], List[float], List[int], List[date], List[datetime]]]]

Source code in synapseclient/models/folder.py

@dataclass()
@async_to_sync
class Folder(FolderSynchronousProtocol, AccessControllable, StorableContainer):
    """Folder is a hierarchical container for organizing data in Synapse.

    Attributes:
        id: The unique immutable ID for this folder. A new ID will be generated for new
            Folders. Once issued, this ID is guaranteed to never change or be re-issued.
        name: The name of this folder. Must be 256 characters or less. Names may only
            contain: letters, numbers, spaces, underscores, hyphens, periods, plus
            signs, apostrophes, and parentheses.
        parent_id: The ID of the Project or Folder that is the parent of this Folder.
        description: The description of this entity. Must be 1000 characters or less.
        etag: (Read Only)
            Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
            concurrent updates. Since the E-Tag changes every time an entity is updated
            it is used to detect when a client's current representation of an entity is
            out-of-date.
        created_on: (Read Only) The date this entity was created.
        modified_on: (Read Only) The date this entity was last modified.
        created_by: (Read Only) The ID of the user that created this entity.
        modified_by: (Read Only) The ID of the user that last modified this entity.
        files: Files that exist within this folder.
        folders: Folders that exist within this folder.
        annotations: Additional metadata associated with the folder. The key is the name
            of your desired annotations. The value is an object containing a list of
            values (use empty list to represent no values for key) and the value type
            associated with all values in the list.  To remove all annotations set this
            to an empty dict `{}`.
    """

    id: Optional[str] = None
    """The unique immutable ID for this folder. A new ID will be generated for new
    Folders. Once issued, this ID is guaranteed to never change or be re-issued"""

    name: Optional[str] = None
    """The name of this folder. Must be 256 characters or less. Names may only contain:
    letters, numbers, spaces, underscores, hyphens, periods, plus signs, apostrophes,
    and parentheses"""

    parent_id: Optional[str] = None
    """The ID of the Project or Folder that is the parent of this Folder."""

    description: Optional[str] = None
    """The description of this entity. Must be 1000 characters or less."""

    etag: Optional[str] = None
    """(Read Only)
    Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
    concurrent updates. Since the E-Tag changes every time an entity is updated it
    is used to detect when a client's current representation of an entity
    is out-of-date."""

    created_on: Optional[str] = None
    """(Read Only) The date this entity was created."""

    modified_on: Optional[str] = None
    """(Read Only) The date this entity was last modified."""

    created_by: Optional[str] = None
    """(Read Only) The ID of the user that created this entity."""

    modified_by: Optional[str] = None
    """(Read Only) The ID of the user that last modified this entity."""

    files: Optional[List["File"]] = field(default_factory=list, compare=False)
    """Files that exist within this folder."""

    folders: Optional[List["Folder"]] = field(default_factory=list, compare=False)
    """Folders that exist within this folder."""

    annotations: Optional[
        Dict[
            str,
            Union[
                List[str],
                List[bool],
                List[float],
                List[int],
                List[date],
                List[datetime],
            ],
        ]
    ] = field(default=None, compare=False)
    """Additional metadata associated with the folder. The key is the name of your
    desired annotations. The value is an object containing a list of values
    (use empty list to represent no values for key) and the value type associated with
    all values in the list. To remove all annotations set this to an empty dict `{}`."""

    is_restricted: bool = field(default=False, repr=False)
    """
    (Store only)

    If set to true, 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.
    You will be contacted with regards to the specific data being restricted and the
    requirements of access.
    """

    create_or_update: bool = field(default=True, repr=False)
    """
    (Store only)

    Indicates whether the method should automatically perform an update if the resource
    conflicts with an existing Synapse object. When True this means that any changes
    to the resource will be non-destructive.

    This boolean is ignored if you've already stored or retrieved the resource from
    Synapse for this instance at least once. Any changes to the resource will be
    destructive in this case. For example if you want to delete the content for a field
    you will need to call `.get()` and then modify the field.
    """

    _last_persistent_instance: Optional["Folder"] = field(
        default=None, repr=False, compare=False
    )
    """The last persistent instance of this object. This is used to determine if the
    object has been changed and needs to be updated in Synapse."""

    @property
    def has_changed(self) -> bool:
        """Determines if the object has been changed and needs to be updated in Synapse."""
        return (
            not self._last_persistent_instance or self._last_persistent_instance != self
        )

    def _set_last_persistent_instance(self) -> None:
        """Stash the last time this object interacted with Synapse. This is used to
        determine if the object has been changed and needs to be updated in Synapse."""
        del self._last_persistent_instance
        self._last_persistent_instance = replace(self)
        self._last_persistent_instance.annotations = (
            deepcopy(self.annotations) if self.annotations else None
        )

    def fill_from_dict(
        self, synapse_folder: Synapse_Folder, set_annotations: bool = True
    ) -> "Folder":
        """
        Converts a response from the REST API into this dataclass.

        Arguments:
            synapse_file: The response from the REST API.
            set_annotations: Whether to set the annotations from the response.

        Returns:
            The Folder object.
        """
        self.id = synapse_folder.get("id", None)
        self.name = synapse_folder.get("name", None)
        self.parent_id = synapse_folder.get("parentId", None)
        self.description = synapse_folder.get("description", None)
        self.etag = synapse_folder.get("etag", None)
        self.created_on = synapse_folder.get("createdOn", None)
        self.modified_on = synapse_folder.get("modifiedOn", None)
        self.created_by = synapse_folder.get("createdBy", None)
        self.modified_by = synapse_folder.get("modifiedBy", None)
        if set_annotations:
            self.annotations = Annotations.from_dict(
                synapse_folder.get("annotations", None)
            )
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Folder_Store: {self.name}"
    )
    async def store_async(
        self,
        parent: Optional[Union["Folder", "Project"]] = None,
        failure_strategy: FailureStrategy = FailureStrategy.LOG_EXCEPTION,
        synapse_client: Optional[Synapse] = None,
    ) -> "Folder":
        """Store folders and files to synapse. If you have any files or folders attached
        to this folder they will be stored as well. You may attach files and folders
        to this folder by setting the `files` and `folders` attributes.

        By default the store operation will non-destructively update the folder if
        you have not already retrieved the folder from Synapse. If you have already
        retrieved the folder from Synapse then the store operation will be destructive
        and will overwrite the folder with the current state of this object. See the
        `create_or_update` attribute for more information.

        Arguments:
            parent: The parent folder or project to store the folder in.
            failure_strategy: Determines how to handle failures when storing attached
                Files and Folders under this Folder and an exception occurs.
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            The folder object.

        Raises:
            ValueError: If the folder does not have an id or a
                (name and (`parent_id` or parent with an id)) set.
        """
        parent_id = parent.id if parent else self.parent_id
        if not (self.id or (self.name and parent_id)):
            raise ValueError(
                "The folder must have an id or a "
                "(name and (`parent_id` or parent with an id)) set."
            )
        self.parent_id = parent_id

        if (
            self.create_or_update
            and not self._last_persistent_instance
            and (
                existing_folder_id := await get_id(
                    entity=self, failure_strategy=None, synapse_client=synapse_client
                )
            )
            and (existing_folder := await Folder(id=existing_folder_id).get_async())
        ):
            merge_dataclass_entities(source=existing_folder, destination=self)

        if self.has_changed:
            loop = asyncio.get_event_loop()
            synapse_folder = Synapse_Folder(
                id=self.id,
                name=self.name,
                parent=parent_id,
                etag=self.etag,
                description=self.description,
            )
            delete_none_keys(synapse_folder)
            current_context = context.get_current()
            entity = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(synapse_client=synapse_client).store(
                        obj=synapse_folder,
                        set_annotations=False,
                        isRestricted=self.is_restricted,
                        createOrUpdate=False,
                    ),
                    current_context,
                ),
            )

            self.fill_from_dict(synapse_folder=entity, set_annotations=False)

        await store_entity_components(
            root_resource=self,
            failure_strategy=failure_strategy,
            synapse_client=synapse_client,
        )
        self._set_last_persistent_instance()
        Synapse.get_client(synapse_client=synapse_client).logger.debug(
            f"Saved Folder {self.name}, id: {self.id}: parent: {self.parent_id}"
        )

        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Folder_Get: {self.id}"
    )
    async def get_async(
        self,
        parent: Optional[Union["Folder", "Project"]] = None,
        synapse_client: Optional[Synapse] = None,
    ) -> "Folder":
        """Get the folder metadata from Synapse. You are able to find a folder by
        either the id or the name and parent_id.

        Arguments:
            parent: The parent folder or project this folder exists under.
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            The folder object.

        Raises:
            ValueError: If the folder does not have an id or a
                (name and (`parent_id` or parent with an id)) set.
        """
        parent_id = parent.id if parent else self.parent_id
        if not (self.id or (self.name and parent_id)):
            raise ValueError(
                "The folder must have an id or a "
                "(name and (`parent_id` or parent with an id)) set."
            )
        self.parent_id = parent_id

        loop = asyncio.get_event_loop()
        current_context = context.get_current()

        entity_id = await get_id(entity=self, synapse_client=synapse_client)

        entity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).get(
                    entity=entity_id,
                ),
                current_context,
            ),
        )

        self.fill_from_dict(synapse_folder=entity, set_annotations=True)

        self._set_last_persistent_instance()
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Folder_Delete: {self.id}"
    )
    async def delete_async(self, synapse_client: Optional[Synapse] = None) -> None:
        """Delete the folder from Synapse by its id.

        Arguments:
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            None

        Raises:
            ValueError: If the folder does not have an id set.
        """
        if not self.id:
            raise ValueError("The folder must have an id set.")
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).delete(
                    obj=self.id,
                ),
                current_context=current_context,
            ),
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Folder_Copy: {self.id}"
    )
    async def copy_async(
        self,
        parent_id: str,
        copy_annotations: bool = True,
        exclude_types: Optional[List[str]] = None,
        file_update_existing: bool = False,
        file_copy_activity: Union[str, None] = "traceback",
        synapse_client: Optional[Synapse] = None,
    ) -> "Folder":
        """
        Copy the folder to another Synapse location. This will recursively copy all
        Tables, Links, Files, and Folders within the folder.

        Arguments:
            parent_id: Synapse ID of a folder/project that the copied entity is
                being copied to
            copy_annotations: True to copy the annotations.
            exclude_types: A list of entity types ['file', 'table', 'link'] which
                determines which entity types to not copy. Defaults to an empty list.
            file_update_existing: When the destination has a file that has the same name,
                users can choose to update that file.
            file_copy_activity: Has three options to set the activity of the copied file:

                    - traceback: Creates a copy of the source files Activity.
                    - existing: Link to the source file's original Activity (if it exists)
                    - None: No activity is set
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            The copied folder object.

        Example: Using this function
            Assuming you have a folder with the ID "syn123" and you want to copy it to a
            project with the ID "syn456":

                new_folder_instance = await Folder(id="syn123").copy_async(parent_id="syn456")

            Copy the folder but do not persist annotations:

                new_folder_instance = await Folder(id="syn123").copy_async(parent_id="syn456", copy_annotations=False)

        Raises:
            ValueError: If the folder does not have an ID and parent_id to copy.
        """
        if not self.id or not parent_id:
            raise ValueError("The folder must have an ID and parent_id to copy.")

        loop = asyncio.get_event_loop()

        current_context = context.get_current()
        syn = Synapse.get_client(synapse_client=synapse_client)
        source_and_destination = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: copy(
                    syn=syn,
                    entity=self.id,
                    destinationId=parent_id,
                    excludeTypes=exclude_types or [],
                    skipCopyAnnotations=not copy_annotations,
                    updateExisting=file_update_existing,
                    setProvenance=file_copy_activity,
                ),
                current_context,
            ),
        )

        new_folder_id = source_and_destination.get(self.id, None)
        if not new_folder_id:
            raise SynapseError("Failed to copy folder.")
        folder_copy = await (
            await Folder(id=new_folder_id).get_async()
        ).sync_from_synapse_async(
            download_file=False,
            synapse_client=synapse_client,
        )
        Synapse.get_client(synapse_client=synapse_client).logger.debug(
            f"Copied from folder {self.id} to {parent_id} with new id of {folder_copy.id}"
        )
        return folder_copy

Functions

get_async(parent=None, synapse_client=None) async

Get the folder metadata from Synapse. You are able to find a folder by either the id or the name and parent_id.

PARAMETER	DESCRIPTION
parent	The parent folder or project this folder exists under. TYPE: Optional[Union[Folder, Project]] DEFAULT: None
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Folder	The folder object.

RAISES	DESCRIPTION
ValueError	If the folder does not have an id or a (name and (parent_id or parent with an id)) set.

Source code in synapseclient/models/folder.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Folder_Get: {self.id}"
)
async def get_async(
    self,
    parent: Optional[Union["Folder", "Project"]] = None,
    synapse_client: Optional[Synapse] = None,
) -> "Folder":
    """Get the folder metadata from Synapse. You are able to find a folder by
    either the id or the name and parent_id.

    Arguments:
        parent: The parent folder or project this folder exists under.
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        The folder object.

    Raises:
        ValueError: If the folder does not have an id or a
            (name and (`parent_id` or parent with an id)) set.
    """
    parent_id = parent.id if parent else self.parent_id
    if not (self.id or (self.name and parent_id)):
        raise ValueError(
            "The folder must have an id or a "
            "(name and (`parent_id` or parent with an id)) set."
        )
    self.parent_id = parent_id

    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    entity_id = await get_id(entity=self, synapse_client=synapse_client)

    entity = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).get(
                entity=entity_id,
            ),
            current_context,
        ),
    )

    self.fill_from_dict(synapse_folder=entity, set_annotations=True)

    self._set_last_persistent_instance()
    return self

store_async(parent=None, failure_strategy=FailureStrategy.LOG_EXCEPTION, synapse_client=None) async

Store folders and files to synapse. If you have any files or folders attached to this folder they will be stored as well. You may attach files and folders to this folder by setting the files and folders attributes.

By default the store operation will non-destructively update the folder if you have not already retrieved the folder from Synapse. If you have already retrieved the folder from Synapse then the store operation will be destructive and will overwrite the folder with the current state of this object. See the create_or_update attribute for more information.

PARAMETER	DESCRIPTION
parent	The parent folder or project to store the folder in. TYPE: Optional[Union[Folder, Project]] DEFAULT: None
failure_strategy	Determines how to handle failures when storing attached Files and Folders under this Folder and an exception occurs. TYPE: FailureStrategy DEFAULT: LOG_EXCEPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Folder	The folder object.

RAISES	DESCRIPTION
ValueError	If the folder does not have an id or a (name and (parent_id or parent with an id)) set.

Source code in synapseclient/models/folder.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Folder_Store: {self.name}"
)
async def store_async(
    self,
    parent: Optional[Union["Folder", "Project"]] = None,
    failure_strategy: FailureStrategy = FailureStrategy.LOG_EXCEPTION,
    synapse_client: Optional[Synapse] = None,
) -> "Folder":
    """Store folders and files to synapse. If you have any files or folders attached
    to this folder they will be stored as well. You may attach files and folders
    to this folder by setting the `files` and `folders` attributes.

    By default the store operation will non-destructively update the folder if
    you have not already retrieved the folder from Synapse. If you have already
    retrieved the folder from Synapse then the store operation will be destructive
    and will overwrite the folder with the current state of this object. See the
    `create_or_update` attribute for more information.

    Arguments:
        parent: The parent folder or project to store the folder in.
        failure_strategy: Determines how to handle failures when storing attached
            Files and Folders under this Folder and an exception occurs.
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        The folder object.

    Raises:
        ValueError: If the folder does not have an id or a
            (name and (`parent_id` or parent with an id)) set.
    """
    parent_id = parent.id if parent else self.parent_id
    if not (self.id or (self.name and parent_id)):
        raise ValueError(
            "The folder must have an id or a "
            "(name and (`parent_id` or parent with an id)) set."
        )
    self.parent_id = parent_id

    if (
        self.create_or_update
        and not self._last_persistent_instance
        and (
            existing_folder_id := await get_id(
                entity=self, failure_strategy=None, synapse_client=synapse_client
            )
        )
        and (existing_folder := await Folder(id=existing_folder_id).get_async())
    ):
        merge_dataclass_entities(source=existing_folder, destination=self)

    if self.has_changed:
        loop = asyncio.get_event_loop()
        synapse_folder = Synapse_Folder(
            id=self.id,
            name=self.name,
            parent=parent_id,
            etag=self.etag,
            description=self.description,
        )
        delete_none_keys(synapse_folder)
        current_context = context.get_current()
        entity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).store(
                    obj=synapse_folder,
                    set_annotations=False,
                    isRestricted=self.is_restricted,
                    createOrUpdate=False,
                ),
                current_context,
            ),
        )

        self.fill_from_dict(synapse_folder=entity, set_annotations=False)

    await store_entity_components(
        root_resource=self,
        failure_strategy=failure_strategy,
        synapse_client=synapse_client,
    )
    self._set_last_persistent_instance()
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Saved Folder {self.name}, id: {self.id}: parent: {self.parent_id}"
    )

    return self

delete_async(synapse_client=None) async

Delete the folder from Synapse by its id.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
None	None

RAISES	DESCRIPTION
ValueError	If the folder does not have an id set.

Source code in synapseclient/models/folder.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Folder_Delete: {self.id}"
)
async def delete_async(self, synapse_client: Optional[Synapse] = None) -> None:
    """Delete the folder from Synapse by its id.

    Arguments:
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        None

    Raises:
        ValueError: If the folder does not have an id set.
    """
    if not self.id:
        raise ValueError("The folder must have an id set.")
    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).delete(
                obj=self.id,
            ),
            current_context=current_context,
        ),
    )

copy_async(parent_id, copy_annotations=True, exclude_types=None, file_update_existing=False, file_copy_activity='traceback', synapse_client=None) async

Copy the folder to another Synapse location. This will recursively copy all Tables, Links, Files, and Folders within the folder.

PARAMETER	DESCRIPTION
parent_id	Synapse ID of a folder/project that the copied entity is being copied to TYPE: str
copy_annotations	True to copy the annotations. TYPE: bool DEFAULT: True
exclude_types	A list of entity types ['file', 'table', 'link'] which determines which entity types to not copy. Defaults to an empty list. TYPE: Optional[List[str]] DEFAULT: None
file_update_existing	When the destination has a file that has the same name, users can choose to update that file. TYPE: bool DEFAULT: False
file_copy_activity	Has three options to set the activity of the copied file: - traceback: Creates a copy of the source files Activity. - existing: Link to the source file's original Activity (if it exists) - None: No activity is set TYPE: Union[str, None] DEFAULT: 'traceback'
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Folder	The copied folder object.

Using this function

Assuming you have a folder with the ID "syn123" and you want to copy it to a project with the ID "syn456":

new_folder_instance = await Folder(id="syn123").copy_async(parent_id="syn456")

Copy the folder but do not persist annotations:

new_folder_instance = await Folder(id="syn123").copy_async(parent_id="syn456", copy_annotations=False)

RAISES	DESCRIPTION
ValueError	If the folder does not have an ID and parent_id to copy.

Source code in synapseclient/models/folder.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Folder_Copy: {self.id}"
)
async def copy_async(
    self,
    parent_id: str,
    copy_annotations: bool = True,
    exclude_types: Optional[List[str]] = None,
    file_update_existing: bool = False,
    file_copy_activity: Union[str, None] = "traceback",
    synapse_client: Optional[Synapse] = None,
) -> "Folder":
    """
    Copy the folder to another Synapse location. This will recursively copy all
    Tables, Links, Files, and Folders within the folder.

    Arguments:
        parent_id: Synapse ID of a folder/project that the copied entity is
            being copied to
        copy_annotations: True to copy the annotations.
        exclude_types: A list of entity types ['file', 'table', 'link'] which
            determines which entity types to not copy. Defaults to an empty list.
        file_update_existing: When the destination has a file that has the same name,
            users can choose to update that file.
        file_copy_activity: Has three options to set the activity of the copied file:

                - traceback: Creates a copy of the source files Activity.
                - existing: Link to the source file's original Activity (if it exists)
                - None: No activity is set
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        The copied folder object.

    Example: Using this function
        Assuming you have a folder with the ID "syn123" and you want to copy it to a
        project with the ID "syn456":

            new_folder_instance = await Folder(id="syn123").copy_async(parent_id="syn456")

        Copy the folder but do not persist annotations:

            new_folder_instance = await Folder(id="syn123").copy_async(parent_id="syn456", copy_annotations=False)

    Raises:
        ValueError: If the folder does not have an ID and parent_id to copy.
    """
    if not self.id or not parent_id:
        raise ValueError("The folder must have an ID and parent_id to copy.")

    loop = asyncio.get_event_loop()

    current_context = context.get_current()
    syn = Synapse.get_client(synapse_client=synapse_client)
    source_and_destination = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: copy(
                syn=syn,
                entity=self.id,
                destinationId=parent_id,
                excludeTypes=exclude_types or [],
                skipCopyAnnotations=not copy_annotations,
                updateExisting=file_update_existing,
                setProvenance=file_copy_activity,
            ),
            current_context,
        ),
    )

    new_folder_id = source_and_destination.get(self.id, None)
    if not new_folder_id:
        raise SynapseError("Failed to copy folder.")
    folder_copy = await (
        await Folder(id=new_folder_id).get_async()
    ).sync_from_synapse_async(
        download_file=False,
        synapse_client=synapse_client,
    )
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Copied from folder {self.id} to {parent_id} with new id of {folder_copy.id}"
    )
    return folder_copy

sync_from_synapse_async(path=None, recursive=True, download_file=True, if_collision=COLLISION_OVERWRITE_LOCAL, failure_strategy=FailureStrategy.LOG_EXCEPTION, synapse_client=None) async

Sync this container and all possible sub-folders from Synapse. By default this will download the files that are found and it will populate the files and folders attributes with the found files and folders. If you only want to retrieve the full tree of metadata about your container specify download_file as False.

This works similar to synapseutils.syncFromSynapse, however, this does not currently support the writing of data to a manifest TSV file. This will be a future enhancement.

Only Files and Folders are supported at this time to be synced from synapse.

PARAMETER	DESCRIPTION
path	An optional path where the file hierarchy will be reproduced. If not specified the files will by default be placed in the synapseCache. TYPE: Optional[str] DEFAULT: None
recursive	Whether or not to recursively get the entire hierarchy of the folder and sub-folders. TYPE: bool DEFAULT: True
download_file	Whether to download the files found or not. TYPE: bool DEFAULT: True
if_collision	Determines how to handle file collisions. May be overwrite.local keep.local keep.both TYPE: str DEFAULT: COLLISION_OVERWRITE_LOCAL
failure_strategy	Determines how to handle failures when retrieving children under this Folder and an exception occurs. TYPE: FailureStrategy DEFAULT: LOG_EXCEPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Self	The object that was called on. This will be the same object that was called on to start the sync.

Using this function

Suppose I want to walk the immediate children of a folder without downloading the files:

from synapseclient import Synapse
from synapseclient.models import Folder

syn = Synapse()
syn.login()

my_folder = Folder(id="syn12345")
await my_folder.sync_from_synapse_async(download_file=False, recursive=False)

for folder in my_folder.folders:
    print(folder.name)

for file in my_folder.files:
    print(file.name)

Suppose I want to download the immediate children of a folder:

from synapseclient import Synapse
from synapseclient.models import Folder

syn = Synapse()
syn.login()

my_folder = Folder(id="syn12345")
await my_folder.sync_from_synapse_async(path="/path/to/folder", recursive=False)

for folder in my_folder.folders:
    print(folder.name)

for file in my_folder.files:
    print(file.name)

Suppose I want to download the immediate all children of a Project and all sub-folders and files:

from synapseclient import Synapse
from synapseclient.models import Project

syn = Synapse()
syn.login()

my_project = Project(id="syn12345")
await my_project.sync_from_synapse_async(path="/path/to/folder")

RAISES	DESCRIPTION
ValueError	If the folder does not have an id set.

A sequence diagram for this method is as follows:

sequenceDiagram
    autonumber
    participant project_or_folder
    activate project_or_folder
    project_or_folder->>sync_from_synapse: Recursive search and download files
    activate sync_from_synapse
        opt Current instance not retrieved from Synapse
            sync_from_synapse->>project_or_folder: Call `.get()` method
            project_or_folder-->>sync_from_synapse: .
        end

        loop For each return of the generator
            sync_from_synapse->>client: call `.getChildren()` method
            client-->>sync_from_synapse: .
            note over sync_from_synapse: Append to a running list
        end

        loop For each child
            note over sync_from_synapse: Create all `pending_tasks` at current depth

            alt Child is File
                note over sync_from_synapse: Append `file.get()` method
            else Child is Folder
                note over sync_from_synapse: Append `folder.get()` method
                alt Recursive is True
                    note over sync_from_synapse: Append `folder.sync_from_synapse()` method
                end
            end
        end

        loop For each task in pending_tasks
            par `file.get()`
                sync_from_synapse->>File: Retrieve File metadata and Optionally download
                File->>client: `.get()`
                client-->>File: .
                File-->>sync_from_synapse: .
            and `folder.get()`
                sync_from_synapse->>Folder: Retrieve Folder metadataa
                Folder->>client: `.get()`
                client-->>Folder: .
                Folder-->>sync_from_synapse: .
            and `folder.sync_from_synapse_async()`
                note over sync_from_synapse: This is a recursive call to `sync_from_synapse`
                sync_from_synapse->>sync_from_synapse: Recursive call to `.sync_from_synapse_async()`
            end
        end

    deactivate sync_from_synapse
    deactivate project_or_folder

Source code in synapseclient/models/mixins/storable_container.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"{self.__class__.__name__}_sync_from_synapse: {self.id}"
)
async def sync_from_synapse_async(
    self: Self,
    path: Optional[str] = None,
    recursive: bool = True,
    download_file: bool = True,
    if_collision: str = COLLISION_OVERWRITE_LOCAL,
    failure_strategy: FailureStrategy = FailureStrategy.LOG_EXCEPTION,
    synapse_client: Optional[Synapse] = None,
) -> Self:
    """
    Sync this container and all possible sub-folders from Synapse. By default this
    will download the files that are found and it will populate the
    `files` and `folders` attributes with the found files and folders. If you only
    want to retrieve the full tree of metadata about your container specify
    `download_file` as False.

    This works similar to [synapseutils.syncFromSynapse][], however, this does not
    currently support the writing of data to a manifest TSV file. This will be a
    future enhancement.

    Only Files and Folders are supported at this time to be synced from synapse.

    Arguments:
        path: An optional path where the file hierarchy will be reproduced. If not
            specified the files will by default be placed in the synapseCache.
        recursive: Whether or not to recursively get the entire hierarchy of the
            folder and sub-folders.
        download_file: Whether to download the files found or not.
        if_collision: Determines how to handle file collisions. May be

            - `overwrite.local`
            - `keep.local`
            - `keep.both`
        failure_strategy: Determines how to handle failures when retrieving children
            under this Folder and an exception occurs.
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        The object that was called on. This will be the same object that was called on
            to start the sync.

    Example: Using this function
        Suppose I want to walk the immediate children of a folder without downloading the files:

            from synapseclient import Synapse
            from synapseclient.models import Folder

            syn = Synapse()
            syn.login()

            my_folder = Folder(id="syn12345")
            await my_folder.sync_from_synapse_async(download_file=False, recursive=False)

            for folder in my_folder.folders:
                print(folder.name)

            for file in my_folder.files:
                print(file.name)

        Suppose I want to download the immediate children of a folder:

            from synapseclient import Synapse
            from synapseclient.models import Folder

            syn = Synapse()
            syn.login()

            my_folder = Folder(id="syn12345")
            await my_folder.sync_from_synapse_async(path="/path/to/folder", recursive=False)

            for folder in my_folder.folders:
                print(folder.name)

            for file in my_folder.files:
                print(file.name)


        Suppose I want to download the immediate all children of a Project and all sub-folders and files:

            from synapseclient import Synapse
            from synapseclient.models import Project

            syn = Synapse()
            syn.login()

            my_project = Project(id="syn12345")
            await my_project.sync_from_synapse_async(path="/path/to/folder")


    Raises:
        ValueError: If the folder does not have an id set.


    A sequence diagram for this method is as follows:

    ```mermaid
    sequenceDiagram
        autonumber
        participant project_or_folder
        activate project_or_folder
        project_or_folder->>sync_from_synapse: Recursive search and download files
        activate sync_from_synapse
            opt Current instance not retrieved from Synapse
                sync_from_synapse->>project_or_folder: Call `.get()` method
                project_or_folder-->>sync_from_synapse: .
            end

            loop For each return of the generator
                sync_from_synapse->>client: call `.getChildren()` method
                client-->>sync_from_synapse: .
                note over sync_from_synapse: Append to a running list
            end

            loop For each child
                note over sync_from_synapse: Create all `pending_tasks` at current depth

                alt Child is File
                    note over sync_from_synapse: Append `file.get()` method
                else Child is Folder
                    note over sync_from_synapse: Append `folder.get()` method
                    alt Recursive is True
                        note over sync_from_synapse: Append `folder.sync_from_synapse()` method
                    end
                end
            end

            loop For each task in pending_tasks
                par `file.get()`
                    sync_from_synapse->>File: Retrieve File metadata and Optionally download
                    File->>client: `.get()`
                    client-->>File: .
                    File-->>sync_from_synapse: .
                and `folder.get()`
                    sync_from_synapse->>Folder: Retrieve Folder metadataa
                    Folder->>client: `.get()`
                    client-->>Folder: .
                    Folder-->>sync_from_synapse: .
                and `folder.sync_from_synapse_async()`
                    note over sync_from_synapse: This is a recursive call to `sync_from_synapse`
                    sync_from_synapse->>sync_from_synapse: Recursive call to `.sync_from_synapse_async()`
                end
            end

        deactivate sync_from_synapse
        deactivate project_or_folder
    ```

    """
    if not self._last_persistent_instance:
        await self.get_async(synapse_client=synapse_client)
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Syncing {self.__class__.__name__} ({self.id}) from Synapse."
    )
    path = os.path.expanduser(path) if path else None

    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    children = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: self._retrieve_children(
                synapse_client=synapse_client,
            ),
            current_context,
        ),
    )

    pending_tasks = []
    self.folders = []
    self.files = []

    for child in children:
        pending_tasks.extend(
            self._create_task_for_child(
                child=child,
                recursive=recursive,
                path=path,
                download_file=download_file,
                if_collision=if_collision,
                failure_strategy=failure_strategy,
                synapse_client=synapse_client,
            )
        )

    for task in asyncio.as_completed(pending_tasks):
        result = await task
        self._resolve_sync_from_synapse_result(
            result=result,
            failure_strategy=failure_strategy,
            synapse_client=synapse_client,
        )
    return self

get_permissions_async(synapse_client=None) async

Get the permissions that the caller has on an Entity.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Permissions	A Permissions object

Using this function:

Getting permissions for a Synapse Entity

permissions = await File(id="syn123").get_permissions_async()

Getting access types list from the Permissions object

permissions.access_types

Source code in synapseclient/models/mixins/access_control.py

async def get_permissions_async(
    self,
    synapse_client: Optional[Synapse] = None,
) -> "Permissions":
    """
    Get the [permissions][synapseclient.core.models.permission.Permissions]
    that the caller has on an Entity.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        A Permissions object


    Example: Using this function:
        Getting permissions for a Synapse Entity

            permissions = await File(id="syn123").get_permissions_async()

        Getting access types list from the Permissions object

            permissions.access_types
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).get_permissions(entity=self.id),
            current_context,
        ),
    )

get_acl_async(principal_id=None, synapse_client=None) async

Get the ACL that a user or group has on an Entity.

PARAMETER	DESCRIPTION
principal_id	Identifier of a user or group (defaults to PUBLIC users) TYPE: int DEFAULT: None
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	An array containing some combination of ['READ', 'UPDATE', 'CREATE', 'DELETE', 'DOWNLOAD', 'MODERATE', 'CHANGE_PERMISSIONS', 'CHANGE_SETTINGS'] or an empty array

Source code in synapseclient/models/mixins/access_control.py

async def get_acl_async(
    self, principal_id: int = None, synapse_client: Optional[Synapse] = None
) -> List[str]:
    """
    Get the [ACL][synapseclient.core.models.permission.Permissions.access_types]
    that a user or group has on an Entity.

    Arguments:
        principal_id: Identifier of a user or group (defaults to PUBLIC users)
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        An array containing some combination of
            ['READ', 'UPDATE', 'CREATE', 'DELETE', 'DOWNLOAD', 'MODERATE',
            'CHANGE_PERMISSIONS', 'CHANGE_SETTINGS']
            or an empty array
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).get_acl(
                entity=self.id, principal_id=principal_id
            ),
            current_context,
        ),
    )

set_permissions_async(principal_id=None, access_type=None, modify_benefactor=False, warn_if_inherits=True, overwrite=True, synapse_client=None) async

Sets permission that a user or group has on an Entity. An Entity may have its own ACL or inherit its ACL from a benefactor.

PARAMETER	DESCRIPTION
principal_id	Identifier of a user or group. 273948 is for all registered Synapse users and 273949 is for public access. None implies public access. TYPE: int DEFAULT: None
access_type	Type of permission to be granted. One or more of CREATE, READ, DOWNLOAD, UPDATE, DELETE, CHANGE_PERMISSIONS. Defaults to ['READ', 'DOWNLOAD'] TYPE: List[str] DEFAULT: None
modify_benefactor	Set as True when modifying a benefactor's ACL TYPE: bool DEFAULT: False
warn_if_inherits	Set as False, when creating a new ACL. Trying to modify the ACL of an Entity that inherits its ACL will result in a warning TYPE: bool DEFAULT: True
overwrite	By default this function overwrites existing permissions for the specified user. Set this flag to False to add new permissions non-destructively. TYPE: bool DEFAULT: True
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, list]]	An Access Control List object

Setting permissions

Grant all registered users download access

await File(id="syn123").set_permissions_async(principal_id=273948, access_type=['READ','DOWNLOAD'])

Grant the public view access

await File(id="syn123").set_permissions_async(principal_id=273949, access_type=['READ'])

Source code in synapseclient/models/mixins/access_control.py

async def set_permissions_async(
    self,
    principal_id: int = None,
    access_type: List[str] = None,
    modify_benefactor: bool = False,
    warn_if_inherits: bool = True,
    overwrite: bool = True,
    synapse_client: Optional[Synapse] = None,
) -> Dict[str, Union[str, list]]:
    """
    Sets permission that a user or group has on an Entity.
    An Entity may have its own ACL or inherit its ACL from a benefactor.

    Arguments:
        principal_id: Identifier of a user or group. `273948` is for all
            registered Synapse users and `273949` is for public access.
            None implies public access.
        access_type: Type of permission to be granted. One or more of CREATE,
            READ, DOWNLOAD, UPDATE, DELETE, CHANGE_PERMISSIONS.

            **Defaults to ['READ', 'DOWNLOAD']**
        modify_benefactor: Set as True when modifying a benefactor's ACL
        warn_if_inherits: Set as False, when creating a new ACL. Trying to modify
            the ACL of an Entity that inherits its ACL will result in a warning
        overwrite: By default this function overwrites existing permissions for
            the specified user. Set this flag to False to add new permissions
            non-destructively.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        An Access Control List object

    Example: Setting permissions
        Grant all registered users download access

            await File(id="syn123").set_permissions_async(principal_id=273948, access_type=['READ','DOWNLOAD'])

        Grant the public view access

            await File(id="syn123").set_permissions_async(principal_id=273949, access_type=['READ'])
    """
    if access_type is None:
        access_type = ["READ", "DOWNLOAD"]
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).setPermissions(
                entity=self.id,
                principalId=principal_id,
                accessType=access_type,
                modify_benefactor=modify_benefactor,
                warn_if_inherits=warn_if_inherits,
                overwrite=overwrite,
            ),
            current_context,
        ),
    )

---

synapseclient.models.File dataclass

Bases: FileSynchronousProtocol, AccessControllable

A file within Synapse.

ATTRIBUTE	DESCRIPTION
id	The unique immutable ID for this file. A new ID will be generated for new Files. Once issued, this ID is guaranteed to never change or be re-issued. TYPE: Optional[str]
name	The name of this entity. Must be 256 characters or less. Names may only contain: letters, numbers, spaces, underscores, hyphens, periods, plus signs, apostrophes, and parentheses. If not specified, the name will be derived from the file name. TYPE: Optional[str]
path	The path to the file on disk. TYPE: Optional[str]
content_type	Used to manually specify Content-type header, for example 'application/png' or 'application/json; charset=UTF-8'. If not specified, the content type will be derived from the file extension. (Create Only) This can be specified only during the initial store of this file. In order to change this after the File has been created use synapseclient.models.File.change_metadata. TYPE: Optional[str]
description	The description of this file. Must be 1000 characters or less. TYPE: Optional[str]
etag	(Read Only) Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle concurrent updates. Since the E-Tag changes every time an entity is updated it is used to detect when a client's current representation of an entity is out-of-date. TYPE: Optional[str]
created_on	(Read Only) The date this entity was created. TYPE: Optional[str]
modified_on	(Read Only) The date this entity was last modified. TYPE: Optional[str]
created_by	(Read Only) The ID of the user that created this entity. TYPE: Optional[str]
modified_by	(Read Only) The ID of the user that last modified this entity. TYPE: Optional[str]
parent_id	The ID of the Entity that is the parent of this Entity. Setting this to a new value and storing it will move this File under the new parent. TYPE: Optional[str]
version_number	(Read Only) The version number issued to this version on the object. TYPE: Optional[int]
version_label	The version label for this entity TYPE: Optional[str]
version_comment	The version comment for this entity TYPE: Optional[str]
is_latest_version	(Read Only) If this is the latest version of the object. TYPE: Optional[bool]
data_file_handle_id	ID of the file associated with this entity. You may define an existing data_file_handle_id to use the existing data_file_handle_id. The creator of the file must also be the owner of the data_file_handle_id to have permission to store the file. TYPE: Optional[str]
file_handle	(Read Only) The file handle associated with this entity. TYPE: Optional[FileHandle]
activity	The Activity model represents the main record of Provenance in Synapse. It is analygous to the Activity defined in the W3C Specification on Provenance. TYPE: Optional[Activity]
annotations	Additional metadata associated with the folder. The key is the name of your desired annotations. The value is an object containing a list of values (use empty list to represent no values for key) and the value type associated with all values in the list. To remove all annotations set this to an empty dict {}. TYPE: Optional[Dict[str, Union[List[str], List[bool], List[float], List[int], List[date], List[datetime]]]]
create_or_update	(Store only) Indicates whether the method should automatically perform an update if the 'obj' conflicts with an existing Synapse object. TYPE: bool
force_version	(Store only) Indicates whether the method should increment the version of the object even if nothing has changed. An update to the MD5 of the file will force a version update regardless of this flag. TYPE: bool
is_restricted	(Store only) If set to true, 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. You will be contacted with regards to the specific data being restricted and the requirements of access. TYPE: bool
synapse_store	(Store only) Whether the File should be uploaded or if false: only the path should be stored when synapseclient.models.File.store is called. TYPE: bool
download_file	(Get only) If True the file will be downloaded. TYPE: bool
download_location	(Get only) The location to download the file to. TYPE: str
if_collision	(Get only) Determines how to handle file collisions. Defaults to "keep.both". May be: overwrite.local keep.local keep.both TYPE: str
synapse_container_limit	(Get only) A Synanpse ID used to limit the search in Synapse if file is specified as a local file. That is, if the file is stored in multiple locations in Synapse only the ones in the specified folder/project will be returned. TYPE: Optional[str]

Source code in synapseclient/models/file.py

@dataclass()
@async_to_sync
class File(FileSynchronousProtocol, AccessControllable):
    """A file within Synapse.

    Attributes:
        id: The unique immutable ID for this file. A new ID will be generated for new
            Files. Once issued, this ID is guaranteed to never change or be re-issued.
        name: The name of this entity. Must be 256 characters or less. Names may only
            contain: letters, numbers, spaces, underscores, hyphens, periods, plus
            signs, apostrophes, and parentheses. If not specified, the name will be
            derived from the file name.
        path: The path to the file on disk.
        content_type: Used to manually specify Content-type header, for example
            'application/png' or 'application/json; charset=UTF-8'. If not specified,
            the content type will be derived from the file extension.


            (Create Only)
            This can be specified only during the initial store of this file.
            In order to change this after the File has been created use
            [synapseclient.models.File.change_metadata][].
        description: The description of this file. Must be 1000 characters or less.
        etag: (Read Only) Synapse employs an Optimistic Concurrency Control (OCC) scheme
            to handle concurrent updates. Since the E-Tag changes every time an entity
            is updated it is used to detect when a client's current representation of an
            entity is out-of-date.
        created_on: (Read Only) The date this entity was created.
        modified_on: (Read Only) The date this entity was last modified.
        created_by: (Read Only) The ID of the user that created this entity.
        modified_by: (Read Only) The ID of the user that last modified this entity.
        parent_id: The ID of the Entity that is the parent of this Entity. Setting this
            to a new value and storing it will move this File under the new parent.
        version_number: (Read Only) The version number issued to this version on the
            object.
        version_label: The version label for this entity
        version_comment: The version comment for this entity
        is_latest_version: (Read Only) If this is the latest version of the object.
        data_file_handle_id: ID of the file associated with this entity. You may define
            an existing data_file_handle_id to use the existing data_file_handle_id. The
            creator of the file must also be the owner of the data_file_handle_id to
            have permission to store the file.
        file_handle: (Read Only) The file handle associated with this entity.
        activity: The Activity model represents the main record of Provenance in
            Synapse. It is analygous to the Activity defined in the
            [W3C Specification](https://www.w3.org/TR/prov-n/) on Provenance.
        annotations: Additional metadata associated with the folder. The key is the name
            of your desired annotations. The value is an object containing a list of
            values (use empty list to represent no values for key) and the value type
            associated with all values in the list. To remove all annotations set this
            to an empty dict `{}`.
        create_or_update: (Store only)
            Indicates whether the method should automatically perform an
            update if the 'obj' conflicts with an existing Synapse object.
        force_version: (Store only)
            Indicates whether the method should increment the version of the
            object even if nothing has changed. An update to the MD5 of the file will
            force a version update regardless of this flag.
        is_restricted: (Store only)
            If set to true, 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. You will be contacted with regards to the specific data
            being restricted and the requirements of access.
        synapse_store: (Store only)
            Whether the File should be uploaded or if false: only the path should
            be stored when [synapseclient.models.File.store][] is called.
        download_file: (Get only) If True the file will be downloaded.
        download_location: (Get only) The location to download the file to.
        if_collision: (Get only)
            Determines how to handle file collisions. Defaults to "keep.both". May be:

            - `overwrite.local`
            - `keep.local`
            - `keep.both`
        synapse_container_limit: (Get only)
            A Synanpse ID used to limit the search in Synapse if
            file is specified as a local file. That is, if the file is stored in
            multiple locations in Synapse only the ones in the specified folder/project
            will be returned.
    """

    id: Optional[str] = None
    """The unique immutable ID for this file. A new ID will be generated for new Files.
    Once issued, this ID is guaranteed to never change or be re-issued."""

    name: Optional[str] = None
    """
    The name of this entity. Must be 256 characters or less.
    Names may only contain: letters, numbers, spaces, underscores, hyphens, periods,
    plus signs, apostrophes, and parentheses. If not specified, the name will be
    derived from the file name.
    """

    path: Optional[str] = None
    """The path to the file on disk."""

    content_type: Optional[str] = None
    """
    Used to manually specify Content-type header, for example 'application/png'
    or 'application/json; charset=UTF-8'. If not specified, the content type will be
    derived from the file extension.


    (Create Only)
    This can be specified only during the initial store of this file. In order to change
    this after the File has been created use
    [synapseclient.models.File.change_metadata][].
    """

    description: Optional[str] = None
    """The description of this file. Must be 1000 characters or less."""

    etag: Optional[str] = None
    """
    (Read Only)
    Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
    concurrent updates. Since the E-Tag changes every time an entity is updated it is
    used to detect when a client's current representation of an entity is out-of-date.
    """

    created_on: Optional[str] = None
    """(Read Only) The date this entity was created."""

    modified_on: Optional[str] = None
    """(Read Only) The date this entity was last modified."""

    created_by: Optional[str] = None
    """(Read Only) The ID of the user that created this entity."""

    modified_by: Optional[str] = None
    """(Read Only) The ID of the user that last modified this entity."""

    parent_id: Optional[str] = None
    """The ID of the Entity that is the parent of this Entity. Setting this to a new
    value and storing it will move this File under the new parent."""

    version_number: Optional[int] = None
    """(Read Only) The version number issued to this version on the object."""

    version_label: Optional[str] = None
    """The version label for this entity"""

    version_comment: Optional[str] = None
    """The version comment for this entity"""

    is_latest_version: Optional[bool] = False
    """(Read Only) If this is the latest version of the object."""

    data_file_handle_id: Optional[str] = None
    """
    ID of the file associated with this entity. You may define an existing
    data_file_handle_id to use the existing data_file_handle_id. The creator of the
    file must also be the owner of the data_file_handle_id to have permission to
    store the file.
    """

    file_handle: Optional[FileHandle] = None
    """(Read Only) The file handle associated with this entity."""

    activity: Optional[Activity] = None
    """The Activity model represents the main record of Provenance in Synapse.  It is
    analygous to the Activity defined in the
    [W3C Specification](https://www.w3.org/TR/prov-n/) on Provenance."""

    annotations: Optional[
        Dict[
            str,
            Union[
                List[str],
                List[bool],
                List[float],
                List[int],
                List[date],
                List[datetime],
            ],
        ]
    ] = None
    """Additional metadata associated with the folder. The key is the name of your
    desired annotations. The value is an object containing a list of values
    (use empty list to represent no values for key) and the value type associated with
    all values in the list. To remove all annotations set this to an empty dict `{}`."""

    create_or_update: bool = field(default=True, repr=False)
    """
    (Store only)

    Indicates whether the method should automatically perform an update if the file
    conflicts with an existing Synapse object.
    """

    force_version: bool = field(default=True, repr=False)
    """
    (Store only)

    Indicates whether the method should increment the version of the object even if
    nothing has changed. An update to the MD5 of the file will force a version update
    regardless of this flag.
    """

    is_restricted: bool = field(default=False, repr=False)
    """
    (Store only)

    If set to true, 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.
    You will be contacted with regards to the specific data being restricted and the
    requirements of access.
    """

    synapse_store: bool = field(default=True, repr=False)
    """
    (Store only)

    Whether the File should be uploaded or if false: only the path should be stored when
    [synapseclient.models.File.store][] is called.
    """

    download_file: bool = field(default=True, repr=False)
    """
    (Get only)

    If True the file will be downloaded."""

    download_location: str = field(default=None, repr=False)
    """
    (Get only)

    The location to download the file to."""

    if_collision: str = field(default="keep.both", repr=False)
    """
    (Get only)

    Determines how to handle file collisions. Defaults to "keep.both".
            May be

            - `overwrite.local`
            - `keep.local`
            - `keep.both`
    """

    synapse_container_limit: Optional[str] = field(default=None, repr=False)
    """A Synanpse ID used to limit the search in Synapse if file is specified as a local
    file. That is, if the file is stored in multiple locations in Synapse only the
    ones in the specified folder/project will be returned."""

    _last_persistent_instance: Optional["File"] = field(
        default=None, repr=False, compare=False
    )
    """The last persistent instance of this object. This is used to determine if the
    object has been changed and needs to be updated in Synapse."""

    def _set_last_persistent_instance(self) -> None:
        """Stash the last time this object interacted with Synapse. This is used to
        determine if the object has been changed and needs to be updated in Synapse."""
        del self._last_persistent_instance
        self._last_persistent_instance = dataclasses.replace(self)
        self._last_persistent_instance.activity = (
            dataclasses.replace(self.activity) if self.activity else None
        )
        self._last_persistent_instance.annotations = (
            deepcopy(self.annotations) if self.annotations else None
        )

    def fill_from_dict(
        self, synapse_file: Union[Synapse_File, Dict], set_annotations: bool = True
    ) -> "File":
        """
        Converts a response from the REST API into this dataclass.

        Arguments:
            synapse_file: The response from the REST API.
            set_annotations: Whether to set the annotations from the response.

        Returns:
            The File object.
        """
        self.id = synapse_file.get("id", None)
        self.name = synapse_file.get("name", None)
        self.path = synapse_file.get("path", None)
        self.description = synapse_file.get("description", None)
        self.etag = synapse_file.get("etag", None)
        self.created_on = synapse_file.get("createdOn", None)
        self.modified_on = synapse_file.get("modifiedOn", None)
        self.created_by = synapse_file.get("createdBy", None)
        self.modified_by = synapse_file.get("modifiedBy", None)
        self.parent_id = synapse_file.get("parentId", None)
        self.version_number = synapse_file.get("versionNumber", None)
        self.version_label = synapse_file.get("versionLabel", None)
        self.version_comment = synapse_file.get("versionComment", None)
        self.is_latest_version = synapse_file.get("isLatestVersion", False)
        self.data_file_handle_id = synapse_file.get("dataFileHandleId", None)
        synapse_file_handle = synapse_file.get("_file_handle", None)
        if synapse_file_handle:
            file_handle = self.file_handle or FileHandle()
            self.file_handle = file_handle
            file_handle.id = synapse_file_handle.get("id", None)
            file_handle.etag = synapse_file_handle.get("etag", None)
            file_handle.created_by = synapse_file_handle.get("createdBy", None)
            file_handle.created_on = synapse_file_handle.get("createdOn", None)
            file_handle.modified_on = synapse_file_handle.get("modifiedOn", None)
            file_handle.concrete_type = synapse_file_handle.get("concreteType", None)
            self.content_type = synapse_file_handle.get("contentType", None)
            file_handle.content_type = synapse_file_handle.get("contentType", None)
            file_handle.content_md5 = synapse_file_handle.get("contentMd5", None)
            file_handle.file_name = synapse_file_handle.get("fileName", None)
            file_handle.storage_location_id = synapse_file_handle.get(
                "storageLocationId", None
            )
            file_handle.content_size = synapse_file_handle.get("contentSize", None)
            file_handle.status = synapse_file_handle.get("status", None)
            file_handle.bucket_name = synapse_file_handle.get("bucketName", None)
            file_handle.key = synapse_file_handle.get("key", None)
            file_handle.preview_id = synapse_file_handle.get("previewId", None)
            file_handle.is_preview = synapse_file_handle.get("isPreview", None)
            file_handle.external_url = synapse_file_handle.get("externalURL", None)

        if set_annotations:
            self.annotations = Annotations.from_dict(
                synapse_file.get("annotations", None)
            )
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"File_Store: {self.path if self.path else self.id}"
    )
    async def store_async(
        self,
        parent: Optional[Union["Folder", "Project"]] = None,
        synapse_client: Optional[Synapse] = None,
    ) -> "File":
        """
        Store the file in Synapse. With this method you may:

        - Upload a file into Synapse
        - Update the metadata of a file in Synapse
        - Store a File object in Synapse without updating a file by setting
            `synapse_store` to False.
        - Change the name of a file in Synapse by setting the `name` attribute of the
            File object. Also see the [synapseclient.models.File.change_metadata][]
            method for changing the name of the downloaded file.
        - Moving a file to a new parent by setting the `parent_id` attribute of the
            File object.

        Arguments:
            parent: The parent folder or project to store the file in. May also be
                specified in the File object. If both are provided the parent passed
                into `store` will take precedence.
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            The file object.


        Example: Using this function
            File with the ID `syn123` at path `path/to/file.txt`:

                file_instance = await File(id="syn123", path="path/to/file.txt").store_async()

            File at the path `path/to/file.txt` and a parent folder with the ID `syn456`:

                file_instance = await File(path="path/to/file.txt", parent_id="syn456").store_async()

            File at the path `path/to/file.txt` and a parent folder with the ID `syn456`:

                file_instance = await File(path="path/to/file.txt").store_async(parent=Folder(id="syn456"))

            Rename a file (Does not update the file on disk or the name of the downloaded file):

                file_instance = await File(id="syn123", download_file=False).get_async()
                print(file_instance.name)  ## prints, e.g., "my_file.txt"
                await file_instance.change_metadata_async(name="my_new_name_file.txt")

            Rename a file, and the name of the file as downloaded (Does not update the file on disk):

                file_instance = await File(id="syn123", download_file=False).get_async()
                print(file_instance.name)  ## prints, e.g., "my_file.txt"
                await file_instance.change_metadata_async(name="my_new_name_file.txt", download_as="my_new_name_file.txt")

        """
        if not (
            self.id is not None
            and (self.path is not None or self.data_file_handle_id is not None)
        ) and not (
            self.path is not None
            and (parent.id if parent else self.parent_id) is not None
        ):
            raise ValueError(
                "The file must have an (ID with a (path or `data_file_handle_id`)), or a "
                "(path with a (`parent_id` or parent with an id)) to store."
            )

        loop = asyncio.get_event_loop()
        synapse_file = Synapse_File(
            id=self.id,
            path=self.path,
            description=self.description,
            etag=self.etag,
            name=self.name or (guess_file_name(self.path) if self.path else None),
            parent=parent.id if parent else self.parent_id,
            contentType=self.content_type,
            dataFileHandleId=self.data_file_handle_id,
            synapseStore=self.synapse_store,
            modifiedOn=self.modified_on,
            versionLabel=self.version_label,
            versionNumber=self.version_number,
            versionComment=self.version_comment,
        )
        delete_none_keys(synapse_file)
        current_context = context.get_current()
        entity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).store(
                    obj=synapse_file,
                    createOrUpdate=self.create_or_update,
                    forceVersion=self.force_version,
                    isRestricted=self.is_restricted,
                    set_annotations=False,
                ),
                current_context,
            ),
        )

        self.fill_from_dict(synapse_file=entity, set_annotations=False)

        re_read_required = await store_entity_components(
            root_resource=self, synapse_client=synapse_client
        )
        if re_read_required:
            before_download_file = self.download_file
            self.download_file = False
            await self.get_async(
                synapse_client=synapse_client,
            )
            self.download_file = before_download_file

        self._set_last_persistent_instance()

        Synapse.get_client(synapse_client=synapse_client).logger.debug(
            f"Stored File {self.name}, id: {self.id}: {self.path}"
        )
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"File_Change_Metadata: {self.id}"
    )
    async def change_metadata_async(
        self,
        name: Optional[str] = None,
        download_as: Optional[str] = None,
        content_type: Optional[str] = None,
        synapse_client: Optional[Synapse] = None,
    ) -> "File":
        """
        Change File Entity metadata for properties that are immutable after creation
        through the store method.

        Arguments:
            name: Specify to change the filename of a file as seen on Synapse.
            download_as: Specify filename to change the filename of a filehandle.
            content_type: Specify content type to change the content type of a
                filehandle.
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            The file object.

        Example: Using this function
            Can be used to change the filename, the filename when the file is downloaded, or the file content-type without downloading:

                file_entity = await File(id="syn123", download_file=False).get_async()
                print(os.path.basename(file_entity.path))  ## prints, e.g., "my_file.txt"
                file_entity = await file_entity.change_metadata_async(name="my_new_name_file.txt", download_as="my_new_downloadAs_name_file.txt", content_type="text/plain")
                print(os.path.basename(file_entity.path))  ## prints, "my_new_downloadAs_name_file.txt"
                print(file_entity.name) ## prints, "my_new_name_file.txt"

        Raises:
            ValueError: If the file does not have an ID to change metadata.
        """
        if not self.id:
            raise ValueError("The file must have an ID to change metadata.")

        loop = asyncio.get_event_loop()

        current_context = context.get_current()
        syn = Synapse.get_client(synapse_client=synapse_client)
        entity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: changeFileMetaData(
                    syn=syn,
                    entity=self.id,
                    name=name,
                    downloadAs=download_as,
                    contentType=content_type,
                    forceVersion=self.force_version,
                ),
                current_context,
            ),
        )

        self.fill_from_dict(synapse_file=entity, set_annotations=True)
        self._set_last_persistent_instance()
        Synapse.get_client(synapse_client=synapse_client).logger.debug(
            f"Change metadata for file {self.name}, id: {self.id}: {self.path}"
        )
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"File_Get: {self.id}, {self.path}"
    )
    async def get_async(
        self,
        include_activity: bool = False,
        synapse_client: Optional[Synapse] = None,
    ) -> "File":
        """
        Get the file from Synapse. You may retrieve a File entity by either:

        - id
        - path


        If you specify both, the `id` will take precedence.


        If you specify the `path` and the file is stored in multiple locations in
        Synapse only the first one found will be returned. The other matching files
        will be printed to the console.


        You may also specify a `version_number` to get a specific version of the file.

        Arguments:
            include_activity: If True the activity will be included in the file
                if it exists.
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            The file object.

        Raises:
            ValueError: If the file does not have an ID or path to get.


        Example: Using this function
            Assuming you have a file with the ID "syn123":

                file_instance = await File(id="syn123").get_async()

            Assuming you have a file at the path "path/to/file.txt":

                file_instance = await File(path="path/to/file.txt").get_async()
        """
        if not self.id and not self.path:
            raise ValueError("The file must have an ID or path to get.")

        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        entity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).get(
                    entity=self.id or self.path,
                    version=self.version_number,
                    ifcollision=self.if_collision,
                    limitSearch=self.synapse_container_limit,
                    downloadFile=self.download_file,
                    downloadLocation=self.download_location,
                ),
                current_context,
            ),
        )

        self.fill_from_dict(synapse_file=entity, set_annotations=True)

        if include_activity:
            self.activity = await Activity.from_parent_async(
                parent=self, synapse_client=synapse_client
            )

        self._set_last_persistent_instance()
        Synapse.get_client(synapse_client=synapse_client).logger.debug(
            f"Got file {self.name}, id: {self.id}, path: {self.path}"
        )
        return self

    @classmethod
    async def from_id_async(
        cls,
        synapse_id: str,
        synapse_client: Optional[Synapse] = None,
    ) -> "File":
        """Wrapper for [synapseclient.models.File.get][].

        Arguments:
            synapse_id: The ID of the file in Synapse.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            The file object.

        Example: Using this function
            Assuming you have a file with the ID "syn123":

                file_instance = await File.from_id_async(synapse_id="syn123")
        """
        return await cls(id=synapse_id).get_async(
            synapse_client=synapse_client,
        )

    @classmethod
    async def from_path_async(
        cls,
        path: str,
        synapse_client: Optional[Synapse] = None,
    ) -> "File":
        """Get the file from Synapse. If the path of the file matches multiple files
        within Synapse the first one found will be returned. The other matching
        files will be printed to the console.


        Wrapper for [synapseclient.models.File.get][].

        Arguments:
            path: The path to the file on disk.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            The file object.

        Example: Using this function
            Assuming you have a file at the path "path/to/file.txt":

                file_instance = await File.from_path_async(path="path/to/file.txt")
        """
        return await cls(path=path).get_async(
            synapse_client=synapse_client,
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"File_Delete: {self.id}"
    )
    async def delete_async(
        self,
        version_only: Optional[bool] = False,
        synapse_client: Optional[Synapse] = None,
    ) -> None:
        """
        Delete the file from Synapse using the ID of the file.

        Arguments:
            version_only: If True only the version specified in the `version_number`
                attribute of the file will be deleted. If False the entire file will
                be deleted.
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            None

        Raises:
            ValueError: If the file does not have an ID to delete.
            ValueError: If the file does not have a version number to delete a version,
                and `version_only` is True.

        Example: Using this function
            Assuming you have a file with the ID "syn123":

                await File(id="syn123").delete_async()
        """
        if not self.id:
            raise ValueError("The file must have an ID to delete.")
        if version_only and not self.version_number:
            raise ValueError("The file must have a version number to delete a version.")

        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).delete(
                    obj=self.id,
                    version=self.version_number if version_only else None,
                ),
                current_context,
            ),
        )
        Synapse.get_client(synapse_client=synapse_client).logger.debug(
            f"Deleted file {self.id}"
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"File_Copy: {self.id}"
    )
    async def copy_async(
        self,
        parent_id: str,
        update_existing: bool = False,
        copy_annotations: bool = True,
        copy_activity: Union[str, None] = "traceback",
        synapse_client: Optional[Synapse] = None,
    ) -> "File":
        """
        Copy the file to another Synapse location. Defaults to the latest version of the
        file, or the version_number specified in the instance.

        Arguments:
            parent_id: Synapse ID of a folder/project that the copied entity is being
                copied to
            update_existing: When the destination has a file that has the same name,
                users can choose to update that file.
            copy_annotations: True to copy the annotations.
            copy_activity: Has three options to set the activity of the copied file:

                    - traceback: Creates a copy of the source files Activity.
                    - existing: Link to the source file's original Activity (if it exists)
                    - None: No activity is set
            synapse_client: If not passed in or None this will use the last client from
                the `.login()` method.

        Returns:
            The copied file object.

        Example: Using this function
            Assuming you have a file with the ID "syn123" and you want to copy it to a folder with the ID "syn456":

                new_file_instance = await File(id="syn123").copy_async(parent_id="syn456")

            Copy the file but do not persist annotations or activity:

                new_file_instance = await File(id="syn123").copy_async(parent_id="syn456", copy_annotations=False, copy_activity=None)

        Raises:
            ValueError: If the file does not have an ID and parent_id to copy.
        """
        if not self.id or not parent_id:
            raise ValueError("The file must have an ID and parent_id to copy.")

        loop = asyncio.get_event_loop()

        current_context = context.get_current()
        syn = Synapse.get_client(synapse_client=synapse_client)
        source_and_destination = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: copy(
                    syn=syn,
                    version=self.version_number,
                    entity=self.id,
                    destinationId=parent_id,
                    skipCopyAnnotations=not copy_annotations,
                    updateExisting=update_existing,
                    setProvenance=copy_activity,
                ),
                current_context,
            ),
        )

        parent_id = source_and_destination.get(self.id, None)
        if not parent_id:
            raise SynapseError("Failed to copy file.")
        file_copy = await File(id=parent_id, download_file=False).get_async(
            synapse_client=synapse_client
        )
        file_copy.download_file = True
        Synapse.get_client(synapse_client=synapse_client).logger.debug(
            f"Copied from file {self.id} to {parent_id} with new id of {file_copy.id}"
        )
        return file_copy

Functions

get_async(include_activity=False, synapse_client=None) async

Get the file from Synapse. You may retrieve a File entity by either:

• id
• path

If you specify both, the id will take precedence.

If you specify the path and the file is stored in multiple locations in Synapse only the first one found will be returned. The other matching files will be printed to the console.

You may also specify a version_number to get a specific version of the file.

PARAMETER	DESCRIPTION
include_activity	If True the activity will be included in the file if it exists. TYPE: bool DEFAULT: False
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
File	The file object.

RAISES	DESCRIPTION
ValueError	If the file does not have an ID or path to get.

Using this function

Assuming you have a file with the ID "syn123":

file_instance = await File(id="syn123").get_async()

Assuming you have a file at the path "path/to/file.txt":

file_instance = await File(path="path/to/file.txt").get_async()

Source code in synapseclient/models/file.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"File_Get: {self.id}, {self.path}"
)
async def get_async(
    self,
    include_activity: bool = False,
    synapse_client: Optional[Synapse] = None,
) -> "File":
    """
    Get the file from Synapse. You may retrieve a File entity by either:

    - id
    - path


    If you specify both, the `id` will take precedence.


    If you specify the `path` and the file is stored in multiple locations in
    Synapse only the first one found will be returned. The other matching files
    will be printed to the console.


    You may also specify a `version_number` to get a specific version of the file.

    Arguments:
        include_activity: If True the activity will be included in the file
            if it exists.
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        The file object.

    Raises:
        ValueError: If the file does not have an ID or path to get.


    Example: Using this function
        Assuming you have a file with the ID "syn123":

            file_instance = await File(id="syn123").get_async()

        Assuming you have a file at the path "path/to/file.txt":

            file_instance = await File(path="path/to/file.txt").get_async()
    """
    if not self.id and not self.path:
        raise ValueError("The file must have an ID or path to get.")

    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    entity = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).get(
                entity=self.id or self.path,
                version=self.version_number,
                ifcollision=self.if_collision,
                limitSearch=self.synapse_container_limit,
                downloadFile=self.download_file,
                downloadLocation=self.download_location,
            ),
            current_context,
        ),
    )

    self.fill_from_dict(synapse_file=entity, set_annotations=True)

    if include_activity:
        self.activity = await Activity.from_parent_async(
            parent=self, synapse_client=synapse_client
        )

    self._set_last_persistent_instance()
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Got file {self.name}, id: {self.id}, path: {self.path}"
    )
    return self

store_async(parent=None, synapse_client=None) async

Store the file in Synapse. With this method you may:

• Upload a file into Synapse
• Update the metadata of a file in Synapse
• Store a File object in Synapse without updating a file by setting synapse_store to False.
• Change the name of a file in Synapse by setting the name attribute of the File object. Also see the synapseclient.models.File.change_metadata method for changing the name of the downloaded file.
• Moving a file to a new parent by setting the parent_id attribute of the File object.

PARAMETER	DESCRIPTION
parent	The parent folder or project to store the file in. May also be specified in the File object. If both are provided the parent passed into store will take precedence. TYPE: Optional[Union[Folder, Project]] DEFAULT: None
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
File	The file object.

Using this function

File with the ID syn123 at path path/to/file.txt:

file_instance = await File(id="syn123", path="path/to/file.txt").store_async()

File at the path path/to/file.txt and a parent folder with the ID syn456:

file_instance = await File(path="path/to/file.txt", parent_id="syn456").store_async()

File at the path path/to/file.txt and a parent folder with the ID syn456:

file_instance = await File(path="path/to/file.txt").store_async(parent=Folder(id="syn456"))

Rename a file (Does not update the file on disk or the name of the downloaded file):

file_instance = await File(id="syn123", download_file=False).get_async()
print(file_instance.name)  ## prints, e.g., "my_file.txt"
await file_instance.change_metadata_async(name="my_new_name_file.txt")

Rename a file, and the name of the file as downloaded (Does not update the file on disk):

file_instance = await File(id="syn123", download_file=False).get_async()
print(file_instance.name)  ## prints, e.g., "my_file.txt"
await file_instance.change_metadata_async(name="my_new_name_file.txt", download_as="my_new_name_file.txt")

Source code in synapseclient/models/file.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"File_Store: {self.path if self.path else self.id}"
)
async def store_async(
    self,
    parent: Optional[Union["Folder", "Project"]] = None,
    synapse_client: Optional[Synapse] = None,
) -> "File":
    """
    Store the file in Synapse. With this method you may:

    - Upload a file into Synapse
    - Update the metadata of a file in Synapse
    - Store a File object in Synapse without updating a file by setting
        `synapse_store` to False.
    - Change the name of a file in Synapse by setting the `name` attribute of the
        File object. Also see the [synapseclient.models.File.change_metadata][]
        method for changing the name of the downloaded file.
    - Moving a file to a new parent by setting the `parent_id` attribute of the
        File object.

    Arguments:
        parent: The parent folder or project to store the file in. May also be
            specified in the File object. If both are provided the parent passed
            into `store` will take precedence.
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        The file object.


    Example: Using this function
        File with the ID `syn123` at path `path/to/file.txt`:

            file_instance = await File(id="syn123", path="path/to/file.txt").store_async()

        File at the path `path/to/file.txt` and a parent folder with the ID `syn456`:

            file_instance = await File(path="path/to/file.txt", parent_id="syn456").store_async()

        File at the path `path/to/file.txt` and a parent folder with the ID `syn456`:

            file_instance = await File(path="path/to/file.txt").store_async(parent=Folder(id="syn456"))

        Rename a file (Does not update the file on disk or the name of the downloaded file):

            file_instance = await File(id="syn123", download_file=False).get_async()
            print(file_instance.name)  ## prints, e.g., "my_file.txt"
            await file_instance.change_metadata_async(name="my_new_name_file.txt")

        Rename a file, and the name of the file as downloaded (Does not update the file on disk):

            file_instance = await File(id="syn123", download_file=False).get_async()
            print(file_instance.name)  ## prints, e.g., "my_file.txt"
            await file_instance.change_metadata_async(name="my_new_name_file.txt", download_as="my_new_name_file.txt")

    """
    if not (
        self.id is not None
        and (self.path is not None or self.data_file_handle_id is not None)
    ) and not (
        self.path is not None
        and (parent.id if parent else self.parent_id) is not None
    ):
        raise ValueError(
            "The file must have an (ID with a (path or `data_file_handle_id`)), or a "
            "(path with a (`parent_id` or parent with an id)) to store."
        )

    loop = asyncio.get_event_loop()
    synapse_file = Synapse_File(
        id=self.id,
        path=self.path,
        description=self.description,
        etag=self.etag,
        name=self.name or (guess_file_name(self.path) if self.path else None),
        parent=parent.id if parent else self.parent_id,
        contentType=self.content_type,
        dataFileHandleId=self.data_file_handle_id,
        synapseStore=self.synapse_store,
        modifiedOn=self.modified_on,
        versionLabel=self.version_label,
        versionNumber=self.version_number,
        versionComment=self.version_comment,
    )
    delete_none_keys(synapse_file)
    current_context = context.get_current()
    entity = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).store(
                obj=synapse_file,
                createOrUpdate=self.create_or_update,
                forceVersion=self.force_version,
                isRestricted=self.is_restricted,
                set_annotations=False,
            ),
            current_context,
        ),
    )

    self.fill_from_dict(synapse_file=entity, set_annotations=False)

    re_read_required = await store_entity_components(
        root_resource=self, synapse_client=synapse_client
    )
    if re_read_required:
        before_download_file = self.download_file
        self.download_file = False
        await self.get_async(
            synapse_client=synapse_client,
        )
        self.download_file = before_download_file

    self._set_last_persistent_instance()

    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Stored File {self.name}, id: {self.id}: {self.path}"
    )
    return self

copy_async(parent_id, update_existing=False, copy_annotations=True, copy_activity='traceback', synapse_client=None) async

Copy the file to another Synapse location. Defaults to the latest version of the file, or the version_number specified in the instance.

PARAMETER	DESCRIPTION
parent_id	Synapse ID of a folder/project that the copied entity is being copied to TYPE: str
update_existing	When the destination has a file that has the same name, users can choose to update that file. TYPE: bool DEFAULT: False
copy_annotations	True to copy the annotations. TYPE: bool DEFAULT: True
copy_activity	Has three options to set the activity of the copied file: - traceback: Creates a copy of the source files Activity. - existing: Link to the source file's original Activity (if it exists) - None: No activity is set TYPE: Union[str, None] DEFAULT: 'traceback'
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
File	The copied file object.

Using this function

Assuming you have a file with the ID "syn123" and you want to copy it to a folder with the ID "syn456":

new_file_instance = await File(id="syn123").copy_async(parent_id="syn456")

Copy the file but do not persist annotations or activity:

new_file_instance = await File(id="syn123").copy_async(parent_id="syn456", copy_annotations=False, copy_activity=None)

RAISES	DESCRIPTION
ValueError	If the file does not have an ID and parent_id to copy.

Source code in synapseclient/models/file.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"File_Copy: {self.id}"
)
async def copy_async(
    self,
    parent_id: str,
    update_existing: bool = False,
    copy_annotations: bool = True,
    copy_activity: Union[str, None] = "traceback",
    synapse_client: Optional[Synapse] = None,
) -> "File":
    """
    Copy the file to another Synapse location. Defaults to the latest version of the
    file, or the version_number specified in the instance.

    Arguments:
        parent_id: Synapse ID of a folder/project that the copied entity is being
            copied to
        update_existing: When the destination has a file that has the same name,
            users can choose to update that file.
        copy_annotations: True to copy the annotations.
        copy_activity: Has three options to set the activity of the copied file:

                - traceback: Creates a copy of the source files Activity.
                - existing: Link to the source file's original Activity (if it exists)
                - None: No activity is set
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        The copied file object.

    Example: Using this function
        Assuming you have a file with the ID "syn123" and you want to copy it to a folder with the ID "syn456":

            new_file_instance = await File(id="syn123").copy_async(parent_id="syn456")

        Copy the file but do not persist annotations or activity:

            new_file_instance = await File(id="syn123").copy_async(parent_id="syn456", copy_annotations=False, copy_activity=None)

    Raises:
        ValueError: If the file does not have an ID and parent_id to copy.
    """
    if not self.id or not parent_id:
        raise ValueError("The file must have an ID and parent_id to copy.")

    loop = asyncio.get_event_loop()

    current_context = context.get_current()
    syn = Synapse.get_client(synapse_client=synapse_client)
    source_and_destination = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: copy(
                syn=syn,
                version=self.version_number,
                entity=self.id,
                destinationId=parent_id,
                skipCopyAnnotations=not copy_annotations,
                updateExisting=update_existing,
                setProvenance=copy_activity,
            ),
            current_context,
        ),
    )

    parent_id = source_and_destination.get(self.id, None)
    if not parent_id:
        raise SynapseError("Failed to copy file.")
    file_copy = await File(id=parent_id, download_file=False).get_async(
        synapse_client=synapse_client
    )
    file_copy.download_file = True
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Copied from file {self.id} to {parent_id} with new id of {file_copy.id}"
    )
    return file_copy

delete_async(version_only=False, synapse_client=None) async

Delete the file from Synapse using the ID of the file.

PARAMETER	DESCRIPTION
version_only	If True only the version specified in the version_number attribute of the file will be deleted. If False the entire file will be deleted. TYPE: Optional[bool] DEFAULT: False
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
None	None

RAISES	DESCRIPTION
ValueError	If the file does not have an ID to delete.
ValueError	If the file does not have a version number to delete a version, and version_only is True.

Using this function

Assuming you have a file with the ID "syn123":

await File(id="syn123").delete_async()

Source code in synapseclient/models/file.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"File_Delete: {self.id}"
)
async def delete_async(
    self,
    version_only: Optional[bool] = False,
    synapse_client: Optional[Synapse] = None,
) -> None:
    """
    Delete the file from Synapse using the ID of the file.

    Arguments:
        version_only: If True only the version specified in the `version_number`
            attribute of the file will be deleted. If False the entire file will
            be deleted.
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        None

    Raises:
        ValueError: If the file does not have an ID to delete.
        ValueError: If the file does not have a version number to delete a version,
            and `version_only` is True.

    Example: Using this function
        Assuming you have a file with the ID "syn123":

            await File(id="syn123").delete_async()
    """
    if not self.id:
        raise ValueError("The file must have an ID to delete.")
    if version_only and not self.version_number:
        raise ValueError("The file must have a version number to delete a version.")

    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).delete(
                obj=self.id,
                version=self.version_number if version_only else None,
            ),
            current_context,
        ),
    )
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Deleted file {self.id}"
    )

from_id_async(synapse_id, synapse_client=None) async classmethod

Wrapper for synapseclient.models.File.get.

PARAMETER	DESCRIPTION
synapse_id	The ID of the file in Synapse. TYPE: str
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
File	The file object.

Using this function

Assuming you have a file with the ID "syn123":

file_instance = await File.from_id_async(synapse_id="syn123")

Source code in synapseclient/models/file.py

@classmethod
async def from_id_async(
    cls,
    synapse_id: str,
    synapse_client: Optional[Synapse] = None,
) -> "File":
    """Wrapper for [synapseclient.models.File.get][].

    Arguments:
        synapse_id: The ID of the file in Synapse.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        The file object.

    Example: Using this function
        Assuming you have a file with the ID "syn123":

            file_instance = await File.from_id_async(synapse_id="syn123")
    """
    return await cls(id=synapse_id).get_async(
        synapse_client=synapse_client,
    )

from_path_async(path, synapse_client=None) async classmethod

Get the file from Synapse. If the path of the file matches multiple files within Synapse the first one found will be returned. The other matching files will be printed to the console.

Wrapper for synapseclient.models.File.get.

PARAMETER	DESCRIPTION
path	The path to the file on disk. TYPE: str
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
File	The file object.

Using this function

Assuming you have a file at the path "path/to/file.txt":

file_instance = await File.from_path_async(path="path/to/file.txt")

Source code in synapseclient/models/file.py

@classmethod
async def from_path_async(
    cls,
    path: str,
    synapse_client: Optional[Synapse] = None,
) -> "File":
    """Get the file from Synapse. If the path of the file matches multiple files
    within Synapse the first one found will be returned. The other matching
    files will be printed to the console.


    Wrapper for [synapseclient.models.File.get][].

    Arguments:
        path: The path to the file on disk.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        The file object.

    Example: Using this function
        Assuming you have a file at the path "path/to/file.txt":

            file_instance = await File.from_path_async(path="path/to/file.txt")
    """
    return await cls(path=path).get_async(
        synapse_client=synapse_client,
    )

change_metadata_async(name=None, download_as=None, content_type=None, synapse_client=None) async

Change File Entity metadata for properties that are immutable after creation through the store method.

PARAMETER	DESCRIPTION
name	Specify to change the filename of a file as seen on Synapse. TYPE: Optional[str] DEFAULT: None
download_as	Specify filename to change the filename of a filehandle. TYPE: Optional[str] DEFAULT: None
content_type	Specify content type to change the content type of a filehandle. TYPE: Optional[str] DEFAULT: None
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
File	The file object.

Using this function

Can be used to change the filename, the filename when the file is downloaded, or the file content-type without downloading:

file_entity = await File(id="syn123", download_file=False).get_async()
print(os.path.basename(file_entity.path))  ## prints, e.g., "my_file.txt"
file_entity = await file_entity.change_metadata_async(name="my_new_name_file.txt", download_as="my_new_downloadAs_name_file.txt", content_type="text/plain")
print(os.path.basename(file_entity.path))  ## prints, "my_new_downloadAs_name_file.txt"
print(file_entity.name) ## prints, "my_new_name_file.txt"

RAISES	DESCRIPTION
ValueError	If the file does not have an ID to change metadata.

Source code in synapseclient/models/file.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"File_Change_Metadata: {self.id}"
)
async def change_metadata_async(
    self,
    name: Optional[str] = None,
    download_as: Optional[str] = None,
    content_type: Optional[str] = None,
    synapse_client: Optional[Synapse] = None,
) -> "File":
    """
    Change File Entity metadata for properties that are immutable after creation
    through the store method.

    Arguments:
        name: Specify to change the filename of a file as seen on Synapse.
        download_as: Specify filename to change the filename of a filehandle.
        content_type: Specify content type to change the content type of a
            filehandle.
        synapse_client: If not passed in or None this will use the last client from
            the `.login()` method.

    Returns:
        The file object.

    Example: Using this function
        Can be used to change the filename, the filename when the file is downloaded, or the file content-type without downloading:

            file_entity = await File(id="syn123", download_file=False).get_async()
            print(os.path.basename(file_entity.path))  ## prints, e.g., "my_file.txt"
            file_entity = await file_entity.change_metadata_async(name="my_new_name_file.txt", download_as="my_new_downloadAs_name_file.txt", content_type="text/plain")
            print(os.path.basename(file_entity.path))  ## prints, "my_new_downloadAs_name_file.txt"
            print(file_entity.name) ## prints, "my_new_name_file.txt"

    Raises:
        ValueError: If the file does not have an ID to change metadata.
    """
    if not self.id:
        raise ValueError("The file must have an ID to change metadata.")

    loop = asyncio.get_event_loop()

    current_context = context.get_current()
    syn = Synapse.get_client(synapse_client=synapse_client)
    entity = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: changeFileMetaData(
                syn=syn,
                entity=self.id,
                name=name,
                downloadAs=download_as,
                contentType=content_type,
                forceVersion=self.force_version,
            ),
            current_context,
        ),
    )

    self.fill_from_dict(synapse_file=entity, set_annotations=True)
    self._set_last_persistent_instance()
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Change metadata for file {self.name}, id: {self.id}: {self.path}"
    )
    return self

get_permissions_async(synapse_client=None) async

Get the permissions that the caller has on an Entity.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Permissions	A Permissions object

Using this function:

Getting permissions for a Synapse Entity

permissions = await File(id="syn123").get_permissions_async()

Getting access types list from the Permissions object

permissions.access_types

Source code in synapseclient/models/mixins/access_control.py

async def get_permissions_async(
    self,
    synapse_client: Optional[Synapse] = None,
) -> "Permissions":
    """
    Get the [permissions][synapseclient.core.models.permission.Permissions]
    that the caller has on an Entity.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        A Permissions object


    Example: Using this function:
        Getting permissions for a Synapse Entity

            permissions = await File(id="syn123").get_permissions_async()

        Getting access types list from the Permissions object

            permissions.access_types
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).get_permissions(entity=self.id),
            current_context,
        ),
    )

get_acl_async(principal_id=None, synapse_client=None) async

Get the ACL that a user or group has on an Entity.

PARAMETER	DESCRIPTION
principal_id	Identifier of a user or group (defaults to PUBLIC users) TYPE: int DEFAULT: None
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	An array containing some combination of ['READ', 'UPDATE', 'CREATE', 'DELETE', 'DOWNLOAD', 'MODERATE', 'CHANGE_PERMISSIONS', 'CHANGE_SETTINGS'] or an empty array

Source code in synapseclient/models/mixins/access_control.py

async def get_acl_async(
    self, principal_id: int = None, synapse_client: Optional[Synapse] = None
) -> List[str]:
    """
    Get the [ACL][synapseclient.core.models.permission.Permissions.access_types]
    that a user or group has on an Entity.

    Arguments:
        principal_id: Identifier of a user or group (defaults to PUBLIC users)
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        An array containing some combination of
            ['READ', 'UPDATE', 'CREATE', 'DELETE', 'DOWNLOAD', 'MODERATE',
            'CHANGE_PERMISSIONS', 'CHANGE_SETTINGS']
            or an empty array
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).get_acl(
                entity=self.id, principal_id=principal_id
            ),
            current_context,
        ),
    )

set_permissions_async(principal_id=None, access_type=None, modify_benefactor=False, warn_if_inherits=True, overwrite=True, synapse_client=None) async

Sets permission that a user or group has on an Entity. An Entity may have its own ACL or inherit its ACL from a benefactor.

PARAMETER	DESCRIPTION
principal_id	Identifier of a user or group. 273948 is for all registered Synapse users and 273949 is for public access. None implies public access. TYPE: int DEFAULT: None
access_type	Type of permission to be granted. One or more of CREATE, READ, DOWNLOAD, UPDATE, DELETE, CHANGE_PERMISSIONS. Defaults to ['READ', 'DOWNLOAD'] TYPE: List[str] DEFAULT: None
modify_benefactor	Set as True when modifying a benefactor's ACL TYPE: bool DEFAULT: False
warn_if_inherits	Set as False, when creating a new ACL. Trying to modify the ACL of an Entity that inherits its ACL will result in a warning TYPE: bool DEFAULT: True
overwrite	By default this function overwrites existing permissions for the specified user. Set this flag to False to add new permissions non-destructively. TYPE: bool DEFAULT: True
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, list]]	An Access Control List object

Setting permissions

Grant all registered users download access

await File(id="syn123").set_permissions_async(principal_id=273948, access_type=['READ','DOWNLOAD'])

Grant the public view access

await File(id="syn123").set_permissions_async(principal_id=273949, access_type=['READ'])

Source code in synapseclient/models/mixins/access_control.py

async def set_permissions_async(
    self,
    principal_id: int = None,
    access_type: List[str] = None,
    modify_benefactor: bool = False,
    warn_if_inherits: bool = True,
    overwrite: bool = True,
    synapse_client: Optional[Synapse] = None,
) -> Dict[str, Union[str, list]]:
    """
    Sets permission that a user or group has on an Entity.
    An Entity may have its own ACL or inherit its ACL from a benefactor.

    Arguments:
        principal_id: Identifier of a user or group. `273948` is for all
            registered Synapse users and `273949` is for public access.
            None implies public access.
        access_type: Type of permission to be granted. One or more of CREATE,
            READ, DOWNLOAD, UPDATE, DELETE, CHANGE_PERMISSIONS.

            **Defaults to ['READ', 'DOWNLOAD']**
        modify_benefactor: Set as True when modifying a benefactor's ACL
        warn_if_inherits: Set as False, when creating a new ACL. Trying to modify
            the ACL of an Entity that inherits its ACL will result in a warning
        overwrite: By default this function overwrites existing permissions for
            the specified user. Set this flag to False to add new permissions
            non-destructively.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        An Access Control List object

    Example: Setting permissions
        Grant all registered users download access

            await File(id="syn123").set_permissions_async(principal_id=273948, access_type=['READ','DOWNLOAD'])

        Grant the public view access

            await File(id="syn123").set_permissions_async(principal_id=273949, access_type=['READ'])
    """
    if access_type is None:
        access_type = ["READ", "DOWNLOAD"]
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).setPermissions(
                entity=self.id,
                principalId=principal_id,
                accessType=access_type,
                modify_benefactor=modify_benefactor,
                warn_if_inherits=warn_if_inherits,
                overwrite=overwrite,
            ),
            current_context,
        ),
    )

---

synapseclient.models.Table dataclass

Bases: TableSynchronousProtocol, AccessControllable

A Table represents the metadata of a table.

ATTRIBUTE	DESCRIPTION
id	The unique immutable ID for this table. A new ID will be generated for new Tables. Once issued, this ID is guaranteed to never change or be re-issued TYPE: Optional[str]
name	The name of this table. Must be 256 characters or less. Names may only contain: letters, numbers, spaces, underscores, hyphens, periods, plus signs, apostrophes, and parentheses TYPE: Optional[str]
parent_id	The ID of the Entity that is the parent of this table. TYPE: Optional[str]
columns	The columns of this table. TYPE: Optional[List[Column]]
description	The description of this entity. Must be 1000 characters or less. TYPE: Optional[List[Column]]
etag	Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle concurrent updates. Since the E-Tag changes every time an entity is updated it is used to detect when a client's current representation of an entity is out-of-date. TYPE: Optional[str]
created_on	The date this table was created. TYPE: Optional[str]
created_by	The ID of the user that created this table. TYPE: Optional[str]
modified_on	The date this table was last modified. In YYYY-MM-DD-Thh:mm:ss.sssZ format TYPE: Optional[str]
modified_by	The ID of the user that last modified this table. TYPE: Optional[str]
version_number	The version number issued to this version on the object. TYPE: Optional[int]
version_label	The version label for this table TYPE: Optional[str]
version_comment	The version comment for this table TYPE: Optional[str]
is_latest_version	If this is the latest version of the object. TYPE: Optional[bool]
is_search_enabled	When creating or updating a table or view specifies if full text search should be enabled. Note that enabling full text search might slow down the indexing of the table or view. TYPE: Optional[bool]
annotations	Additional metadata associated with the table. The key is the name of your desired annotations. The value is an object containing a list of values (use empty list to represent no values for key) and the value type associated with all values in the list. TYPE: Optional[Dict[str, Union[List[str], List[bool], List[float], List[int], List[date], List[datetime]]]]

Source code in synapseclient/models/table.py

@dataclass()
@async_to_sync
class Table(TableSynchronousProtocol, AccessControllable):
    """A Table represents the metadata of a table.

    Attributes:
        id: The unique immutable ID for this table. A new ID will be generated for new
            Tables. Once issued, this ID is guaranteed to never change or be re-issued
        name: The name of this table. Must be 256 characters or less. Names may only
            contain: letters, numbers, spaces, underscores, hyphens, periods, plus
            signs, apostrophes, and parentheses
        parent_id: The ID of the Entity that is the parent of this table.
        columns: The columns of this table.
        description: The description of this entity. Must be 1000 characters or less.
        etag: Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
            concurrent updates. Since the E-Tag changes every time an entity is updated
            it is used to detect when a client's current representation of an entity is
            out-of-date.
        created_on: The date this table was created.
        created_by: The ID of the user that created this table.
        modified_on: The date this table was last modified.
            In YYYY-MM-DD-Thh:mm:ss.sssZ format
        modified_by: The ID of the user that last modified this table.
        version_number: The version number issued to this version on the object.
        version_label: The version label for this table
        version_comment: The version comment for this table
        is_latest_version: If this is the latest version of the object.
        is_search_enabled: When creating or updating a table or view specifies if full
            text search should be enabled. Note that enabling full text search might
            slow down the indexing of the table or view.
        annotations: Additional metadata associated with the table. The key is the
            name of your desired annotations. The value is an object containing a list
            of values (use empty list to represent no values for key) and the value
            type associated with all values in the list.

    """

    id: Optional[str] = None
    """The unique immutable ID for this table. A new ID will be generated for new
    Tables. Once issued, this ID is guaranteed to never change or be re-issued"""

    name: Optional[str] = None
    """The name of this table. Must be 256 characters or less. Names may only
    contain: letters, numbers, spaces, underscores, hyphens, periods, plus signs,
    apostrophes, and parentheses"""

    parent_id: Optional[str] = None
    """The ID of the Entity that is the parent of this table."""

    columns: Optional[List[Column]] = None

    # TODO: Description doesn't seem to be returned from the API. Look into why.
    # description: Optional[str] = None
    # """The description of this entity. Must be 1000 characters or less."""

    etag: Optional[str] = None
    """
    Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
    concurrent updates. Since the E-Tag changes every time an entity is updated it is
    used to detect when a client's current representation of an entity is out-of-date.
    """

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

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

    modified_on: Optional[str] = None
    """The date this table was last modified. In YYYY-MM-DD-Thh:mm:ss.sssZ format"""

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

    version_number: Optional[int] = None
    """The version number issued to this version on the object."""

    version_label: Optional[str] = None
    """The version label for this table"""

    version_comment: Optional[str] = None
    """The version comment for this table"""

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

    is_search_enabled: Optional[bool] = None
    """When creating or updating a table or view specifies if full text search
    should be enabled. Note that enabling full text search might slow down the
    indexing of the table or view."""

    activity: Optional[Activity] = None
    """The Activity model represents the main record of Provenance in Synapse.  It is
    analygous to the Activity defined in the
    [W3C Specification](https://www.w3.org/TR/prov-n/) on Provenance. """

    annotations: Optional[
        Dict[
            str,
            Union[
                List[str],
                List[bool],
                List[float],
                List[int],
                List[date],
                List[datetime],
            ],
        ]
    ] = None
    """Additional metadata associated with the table. The key is the name of your
    desired annotations. The value is an object containing a list of values
    (use empty list to represent no values for key) and the value type associated with
    all values in the list. To remove all annotations set this to an empty dict `{}`."""

    def fill_from_dict(
        self, synapse_table: Synapse_Table, set_annotations: bool = True
    ) -> "Table":
        """Converts the data coming from the Synapse API into this datamodel.

        :param synapse_table: The data coming from the Synapse API
        """
        self.id = synapse_table.get("id", None)
        self.name = synapse_table.get("name", None)
        self.parent_id = synapse_table.get("parentId", None)
        # TODO: Description doesn't seem to be returned from the API. Look into why.
        # self.description = synapse_table.description
        self.etag = synapse_table.get("etag", None)
        self.created_on = synapse_table.get("createdOn", None)
        self.created_by = synapse_table.get("createdBy", None)
        self.modified_on = synapse_table.get("modifiedOn", None)
        self.modified_by = synapse_table.get("modifiedBy", None)
        self.version_number = synapse_table.get("versionNumber", None)
        self.version_label = synapse_table.get("versionLabel", None)
        self.version_comment = synapse_table.get("versionComment", None)
        self.is_latest_version = synapse_table.get("isLatestVersion", None)
        self.is_search_enabled = synapse_table.get("isSearchEnabled", False)
        self.columns = [
            Column(id=columnId, name=None, column_type=None)
            for columnId in synapse_table.get("columnIds", [])
        ]
        if set_annotations:
            self.annotations = Annotations.from_dict(
                synapse_table.get("annotations", None)
            )
        return self

    @otel_trace_method(
        method_to_trace_name=lambda _, **kwargs: f"Store_rows_by_csv: {kwargs.get('csv_path', None)}"
    )
    async def store_rows_from_csv_async(
        self, csv_path: str, synapse_client: Optional[Synapse] = None
    ) -> str:
        """Takes in a path to a CSV and stores the rows to Synapse.

        Arguments:
            csv_path: The path to the CSV to store.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            The path to the CSV that was stored.
        """
        synapse_table = Synapse_Table(schema=self.id, values=csv_path)
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        entity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).store(
                    obj=synapse_table
                ),
                current_context,
            ),
        )
        print(entity)
        # TODO: What should this return?
        return csv_path

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Delete_rows: {self.name}"
    )
    async def delete_rows_async(
        self, rows: List[Row], synapse_client: Optional[Synapse] = None
    ) -> None:
        """Delete rows from a table.

        Arguments:
            rows: The rows to delete.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            None
        """
        rows_to_delete = []
        for row in rows:
            rows_to_delete.append([row.row_id, row.version_number])
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: delete_rows(
                    syn=Synapse.get_client(synapse_client=synapse_client),
                    table_id=self.id,
                    row_id_vers_list=rows_to_delete,
                ),
                current_context,
            ),
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Table_Schema_Store: {self.name}"
    )
    async def store_schema_async(
        self, synapse_client: Optional[Synapse] = None
    ) -> "Table":
        """Store non-row information about a table including the columns and annotations.

        Arguments:
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            The Table instance stored in synapse.
        """
        tasks = []
        if self.columns:
            # TODO: When a table is retrieved via `.get()` we create Column objects but
            # TODO: We only have the ID attribute. THis is causing this if check to eval
            # TODO: To True, however, we aren't actually modifying the column.
            # TODO: Perhaps we should have a `has_changed` boolean on all dataclasses
            # TODO: That we can check to see if we need to store the data.
            tasks.extend(
                column.store_async(synapse_client=synapse_client)
                for column in self.columns
            )
            try:
                results = await asyncio.gather(*tasks, return_exceptions=True)

                # TODO: Proper exception handling
                for result in results:
                    if isinstance(result, Column):
                        print(f"Stored {result.name}")
                    else:
                        if isinstance(result, BaseException):
                            raise result
                        raise ValueError(f"Unknown type: {type(result)}", result)
            except Exception as ex:
                Synapse.get_client(synapse_client=synapse_client).logger.exception(ex)
                print("I hit an exception")

        synapse_schema = Synapse_Schema(
            name=self.name,
            columns=self.columns,
            parent=self.parent_id,
        )

        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        entity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).store(
                    obj=synapse_schema
                ),
                current_context,
            ),
        )

        self.fill_from_dict(synapse_table=entity, set_annotations=False)

        re_read_required = await store_entity_components(
            root_resource=self, synapse_client=synapse_client
        )
        if re_read_required:
            await self.get_async(
                synapse_client=synapse_client,
            )

        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Table_Get: {self.name}"
    )
    async def get_async(self, synapse_client: Optional[Synapse] = None) -> "Table":
        """Get the metadata about the table from synapse.

        Arguments:
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            The Table instance stored in synapse.
        """
        # TODO: How do we want to support retriving the table? Do we want to support by name, and parent?
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        entity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).get(
                    entity=self.id
                ),
                current_context,
            ),
        )
        self.fill_from_dict(synapse_table=entity, set_annotations=True)
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Table_Delete: {self.name}"
    )
    # TODO: Synapse allows immediate deletion of entities, but the Synapse Client does not
    # TODO: Should we support immediate deletion?
    async def delete_async(self, synapse_client: Optional[Synapse] = None) -> None:
        """Delete the table from synapse.

        Arguments:
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            None
        """
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).delete(
                    obj=self.id
                ),
                current_context,
            ),
        )

    @classmethod
    async def query_async(
        cls,
        query: str,
        result_format: Union[CsvResultFormat, RowsetResultFormat] = CsvResultFormat(),
        synapse_client: Optional[Synapse] = None,
    ) -> Union[Synapse_CsvFileTable, Synaspe_TableQueryResult]:
        """Query for data on a table stored in Synapse.

        Arguments:
            query: The query to run.
            result_format: The format of the results. Defaults to CsvResultFormat().
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            The results of the query.
        """
        with tracer.start_as_current_span("Table_query"):
            loop = asyncio.get_event_loop()
            current_context = context.get_current()

            # TODO: Future Idea - We stream back a CSV, and let those reading this to handle the CSV however they want
            results = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(
                        synapse_client=synapse_client
                    ).tableQuery(
                        query=query,
                        **result_format.to_dict(),
                    ),
                    current_context,
                ),
            )
            print(results)
            return results

Functions

get_async(synapse_client=None) async

Get the metadata about the table from synapse.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Table	The Table instance stored in synapse.

Source code in synapseclient/models/table.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Table_Get: {self.name}"
)
async def get_async(self, synapse_client: Optional[Synapse] = None) -> "Table":
    """Get the metadata about the table from synapse.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        The Table instance stored in synapse.
    """
    # TODO: How do we want to support retriving the table? Do we want to support by name, and parent?
    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    entity = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).get(
                entity=self.id
            ),
            current_context,
        ),
    )
    self.fill_from_dict(synapse_table=entity, set_annotations=True)
    return self

store_schema_async(synapse_client=None) async

Store non-row information about a table including the columns and annotations.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Table	The Table instance stored in synapse.

Source code in synapseclient/models/table.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Table_Schema_Store: {self.name}"
)
async def store_schema_async(
    self, synapse_client: Optional[Synapse] = None
) -> "Table":
    """Store non-row information about a table including the columns and annotations.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        The Table instance stored in synapse.
    """
    tasks = []
    if self.columns:
        # TODO: When a table is retrieved via `.get()` we create Column objects but
        # TODO: We only have the ID attribute. THis is causing this if check to eval
        # TODO: To True, however, we aren't actually modifying the column.
        # TODO: Perhaps we should have a `has_changed` boolean on all dataclasses
        # TODO: That we can check to see if we need to store the data.
        tasks.extend(
            column.store_async(synapse_client=synapse_client)
            for column in self.columns
        )
        try:
            results = await asyncio.gather(*tasks, return_exceptions=True)

            # TODO: Proper exception handling
            for result in results:
                if isinstance(result, Column):
                    print(f"Stored {result.name}")
                else:
                    if isinstance(result, BaseException):
                        raise result
                    raise ValueError(f"Unknown type: {type(result)}", result)
        except Exception as ex:
            Synapse.get_client(synapse_client=synapse_client).logger.exception(ex)
            print("I hit an exception")

    synapse_schema = Synapse_Schema(
        name=self.name,
        columns=self.columns,
        parent=self.parent_id,
    )

    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    entity = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).store(
                obj=synapse_schema
            ),
            current_context,
        ),
    )

    self.fill_from_dict(synapse_table=entity, set_annotations=False)

    re_read_required = await store_entity_components(
        root_resource=self, synapse_client=synapse_client
    )
    if re_read_required:
        await self.get_async(
            synapse_client=synapse_client,
        )

    return self

store_rows_from_csv_async(csv_path, synapse_client=None) async

Takes in a path to a CSV and stores the rows to Synapse.

PARAMETER	DESCRIPTION
csv_path	The path to the CSV to store. TYPE: str
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
str	The path to the CSV that was stored.

Source code in synapseclient/models/table.py

@otel_trace_method(
    method_to_trace_name=lambda _, **kwargs: f"Store_rows_by_csv: {kwargs.get('csv_path', None)}"
)
async def store_rows_from_csv_async(
    self, csv_path: str, synapse_client: Optional[Synapse] = None
) -> str:
    """Takes in a path to a CSV and stores the rows to Synapse.

    Arguments:
        csv_path: The path to the CSV to store.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        The path to the CSV that was stored.
    """
    synapse_table = Synapse_Table(schema=self.id, values=csv_path)
    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    entity = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).store(
                obj=synapse_table
            ),
            current_context,
        ),
    )
    print(entity)
    # TODO: What should this return?
    return csv_path

delete_rows_async(rows, synapse_client=None) async

Delete rows from a table.

PARAMETER	DESCRIPTION
rows	The rows to delete. TYPE: List[Row]
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
None	None

Source code in synapseclient/models/table.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Delete_rows: {self.name}"
)
async def delete_rows_async(
    self, rows: List[Row], synapse_client: Optional[Synapse] = None
) -> None:
    """Delete rows from a table.

    Arguments:
        rows: The rows to delete.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        None
    """
    rows_to_delete = []
    for row in rows:
        rows_to_delete.append([row.row_id, row.version_number])
    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: delete_rows(
                syn=Synapse.get_client(synapse_client=synapse_client),
                table_id=self.id,
                row_id_vers_list=rows_to_delete,
            ),
            current_context,
        ),
    )

query_async(query, result_format=CsvResultFormat(), synapse_client=None) async classmethod

Query for data on a table stored in Synapse.

PARAMETER	DESCRIPTION
query	The query to run. TYPE: str
result_format	The format of the results. Defaults to CsvResultFormat(). TYPE: Union[CsvResultFormat, RowsetResultFormat] DEFAULT: CsvResultFormat()
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Union[CsvFileTable, TableQueryResult]	The results of the query.

Source code in synapseclient/models/table.py

@classmethod
async def query_async(
    cls,
    query: str,
    result_format: Union[CsvResultFormat, RowsetResultFormat] = CsvResultFormat(),
    synapse_client: Optional[Synapse] = None,
) -> Union[Synapse_CsvFileTable, Synaspe_TableQueryResult]:
    """Query for data on a table stored in Synapse.

    Arguments:
        query: The query to run.
        result_format: The format of the results. Defaults to CsvResultFormat().
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        The results of the query.
    """
    with tracer.start_as_current_span("Table_query"):
        loop = asyncio.get_event_loop()
        current_context = context.get_current()

        # TODO: Future Idea - We stream back a CSV, and let those reading this to handle the CSV however they want
        results = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).tableQuery(
                    query=query,
                    **result_format.to_dict(),
                ),
                current_context,
            ),
        )
        print(results)
        return results

delete_async(synapse_client=None) async

Delete the table from synapse.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
None	None

Source code in synapseclient/models/table.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Table_Delete: {self.name}"
)
# TODO: Synapse allows immediate deletion of entities, but the Synapse Client does not
# TODO: Should we support immediate deletion?
async def delete_async(self, synapse_client: Optional[Synapse] = None) -> None:
    """Delete the table from synapse.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        None
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).delete(
                obj=self.id
            ),
            current_context,
        ),
    )

get_permissions_async(synapse_client=None) async

Get the permissions that the caller has on an Entity.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Permissions	A Permissions object

Using this function:

Getting permissions for a Synapse Entity

permissions = await File(id="syn123").get_permissions_async()

Getting access types list from the Permissions object

permissions.access_types

Source code in synapseclient/models/mixins/access_control.py

async def get_permissions_async(
    self,
    synapse_client: Optional[Synapse] = None,
) -> "Permissions":
    """
    Get the [permissions][synapseclient.core.models.permission.Permissions]
    that the caller has on an Entity.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        A Permissions object


    Example: Using this function:
        Getting permissions for a Synapse Entity

            permissions = await File(id="syn123").get_permissions_async()

        Getting access types list from the Permissions object

            permissions.access_types
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).get_permissions(entity=self.id),
            current_context,
        ),
    )

get_acl_async(principal_id=None, synapse_client=None) async

Get the ACL that a user or group has on an Entity.

PARAMETER	DESCRIPTION
principal_id	Identifier of a user or group (defaults to PUBLIC users) TYPE: int DEFAULT: None
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	An array containing some combination of ['READ', 'UPDATE', 'CREATE', 'DELETE', 'DOWNLOAD', 'MODERATE', 'CHANGE_PERMISSIONS', 'CHANGE_SETTINGS'] or an empty array

Source code in synapseclient/models/mixins/access_control.py

async def get_acl_async(
    self, principal_id: int = None, synapse_client: Optional[Synapse] = None
) -> List[str]:
    """
    Get the [ACL][synapseclient.core.models.permission.Permissions.access_types]
    that a user or group has on an Entity.

    Arguments:
        principal_id: Identifier of a user or group (defaults to PUBLIC users)
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        An array containing some combination of
            ['READ', 'UPDATE', 'CREATE', 'DELETE', 'DOWNLOAD', 'MODERATE',
            'CHANGE_PERMISSIONS', 'CHANGE_SETTINGS']
            or an empty array
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).get_acl(
                entity=self.id, principal_id=principal_id
            ),
            current_context,
        ),
    )

set_permissions_async(principal_id=None, access_type=None, modify_benefactor=False, warn_if_inherits=True, overwrite=True, synapse_client=None) async

Sets permission that a user or group has on an Entity. An Entity may have its own ACL or inherit its ACL from a benefactor.

PARAMETER	DESCRIPTION
principal_id	Identifier of a user or group. 273948 is for all registered Synapse users and 273949 is for public access. None implies public access. TYPE: int DEFAULT: None
access_type	Type of permission to be granted. One or more of CREATE, READ, DOWNLOAD, UPDATE, DELETE, CHANGE_PERMISSIONS. Defaults to ['READ', 'DOWNLOAD'] TYPE: List[str] DEFAULT: None
modify_benefactor	Set as True when modifying a benefactor's ACL TYPE: bool DEFAULT: False
warn_if_inherits	Set as False, when creating a new ACL. Trying to modify the ACL of an Entity that inherits its ACL will result in a warning TYPE: bool DEFAULT: True
overwrite	By default this function overwrites existing permissions for the specified user. Set this flag to False to add new permissions non-destructively. TYPE: bool DEFAULT: True
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, list]]	An Access Control List object

Setting permissions

Grant all registered users download access

await File(id="syn123").set_permissions_async(principal_id=273948, access_type=['READ','DOWNLOAD'])

Grant the public view access

await File(id="syn123").set_permissions_async(principal_id=273949, access_type=['READ'])

Source code in synapseclient/models/mixins/access_control.py

async def set_permissions_async(
    self,
    principal_id: int = None,
    access_type: List[str] = None,
    modify_benefactor: bool = False,
    warn_if_inherits: bool = True,
    overwrite: bool = True,
    synapse_client: Optional[Synapse] = None,
) -> Dict[str, Union[str, list]]:
    """
    Sets permission that a user or group has on an Entity.
    An Entity may have its own ACL or inherit its ACL from a benefactor.

    Arguments:
        principal_id: Identifier of a user or group. `273948` is for all
            registered Synapse users and `273949` is for public access.
            None implies public access.
        access_type: Type of permission to be granted. One or more of CREATE,
            READ, DOWNLOAD, UPDATE, DELETE, CHANGE_PERMISSIONS.

            **Defaults to ['READ', 'DOWNLOAD']**
        modify_benefactor: Set as True when modifying a benefactor's ACL
        warn_if_inherits: Set as False, when creating a new ACL. Trying to modify
            the ACL of an Entity that inherits its ACL will result in a warning
        overwrite: By default this function overwrites existing permissions for
            the specified user. Set this flag to False to add new permissions
            non-destructively.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        An Access Control List object

    Example: Setting permissions
        Grant all registered users download access

            await File(id="syn123").set_permissions_async(principal_id=273948, access_type=['READ','DOWNLOAD'])

        Grant the public view access

            await File(id="syn123").set_permissions_async(principal_id=273949, access_type=['READ'])
    """
    if access_type is None:
        access_type = ["READ", "DOWNLOAD"]
    loop = asyncio.get_event_loop()
    current_context = context.get_current()

    return await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).setPermissions(
                entity=self.id,
                principalId=principal_id,
                accessType=access_type,
                modify_benefactor=modify_benefactor,
                warn_if_inherits=warn_if_inherits,
                overwrite=overwrite,
            ),
            current_context,
        ),
    )

---

synapseclient.models.Activity dataclass

Bases: ActivitySynchronousProtocol

An activity is a Synapse object that helps to keep track of what objects were used in an analysis step, as well as what objects were generated. Thus, all relationships between Synapse objects and an activity are governed by dependencies. That is, an activity needs to know what it used, and outputs need to know what activity they were generatedBy.

ATTRIBUTE	DESCRIPTION
id	The unique immutable ID for this actvity. TYPE: Optional[str]
name	A name for this Activity. TYPE: Optional[str]
description	A description for this Activity. TYPE: Optional[str]
etag	Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle concurrent updates. Since the E-Tag changes every time an entity is updated it is used to detect when a client's current representation of an entity is out-of-date. TYPE: Optional[str]
created_on	The date this object was created. TYPE: Optional[str]
modified_on	The date this object was last modified. TYPE: Optional[str]
created_by	The user that created this object. TYPE: Optional[str]
modified_by	The user that last modified this object. TYPE: Optional[str]
used	The entities or URLs used by this Activity. TYPE: List[Union[UsedEntity, UsedURL]]
executed	The entities or URLs executed by this Activity. TYPE: List[Union[UsedEntity, UsedURL]]

Source code in synapseclient/models/activity.py

@dataclass
@async_to_sync
class Activity(ActivitySynchronousProtocol):
    """
    An activity is a Synapse object that helps to keep track of what objects were used
    in an analysis step, as well as what objects were generated. Thus, all relationships
    between Synapse objects and an activity are governed by dependencies. That is, an
    activity needs to know what it `used`, and outputs need to know what activity
    they were `generatedBy`.

    Attributes:
        id: The unique immutable ID for this actvity.
        name: A name for this Activity.
        description: A description for this Activity.
        etag: Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
            concurrent updates. Since the E-Tag changes every time an entity is updated
            it is used to detect when a client's current representation of an entity is
            out-of-date.
        created_on: The date this object was created.
        modified_on: The date this object was last modified.
        created_by: The user that created this object.
        modified_by: The user that last modified this object.
        used: The entities or URLs used by this Activity.
        executed: The entities or URLs executed by this Activity.
    """

    id: Optional[str] = None
    """The unique immutable ID for this actvity."""

    name: Optional[str] = None
    """A name for this Activity."""

    description: Optional[str] = None
    """A description for this Activity."""

    etag: Optional[str] = None
    """
    Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
    concurrent updates. Since the E-Tag changes every time an entity is updated it is
    used to detect when a client's current representation of an entity is out-of-date.
    """

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

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

    created_by: Optional[str] = None
    """The user that created this object."""

    modified_by: Optional[str] = None
    """The user that last modified this object."""

    used: List[Union[UsedEntity, UsedURL]] = field(default_factory=list)
    """The entities used by this Activity."""

    executed: List[Union[UsedEntity, UsedURL]] = field(default_factory=list)
    """The entities executed by this Activity."""

    def fill_from_dict(
        self, synapse_activity: Union[Synapse_Activity, Dict]
    ) -> "Activity":
        """
        Converts a response from the REST API into this dataclass.

        Arguments:
            synapse_activity: The response from the REST API.

        Returns:
            The Activity object.
        """
        self.id = synapse_activity.get("id", None)
        self.name = synapse_activity.get("name", None)
        self.description = synapse_activity.get("description", None)
        self.etag = synapse_activity.get("etag", None)
        self.created_on = synapse_activity.get("createdOn", None)
        self.modified_on = synapse_activity.get("modifiedOn", None)
        self.created_by = synapse_activity.get("createdBy", None)
        self.modified_by = synapse_activity.get("modifiedBy", None)
        self.executed = []
        self.used = []
        for used in synapse_activity.get("used", []):
            concrete_type = used.get("concreteType", None)
            if USED_URL == concrete_type:
                used_url = UsedURL(
                    name=used.get("name", None),
                    url=used.get("url", None),
                )

                if used.get("wasExecuted", False):
                    self.executed.append(used_url)
                else:
                    self.used.append(used_url)
            elif USED_ENTITY == concrete_type:
                reference = used.get("reference", {})
                used_entity = UsedEntity(
                    target_id=reference.get("targetId", None),
                    target_version_number=reference.get("targetVersionNumber", None),
                )

                if used.get("wasExecuted", False):
                    self.executed.append(used_entity)
                else:
                    self.used.append(used_entity)

        return self

    def _create_used_and_executed_synapse_activities(
        self,
    ) -> UsedAndExecutedSynapseActivities:
        """
        Helper function to create the used and executed activities for the
        Synapse Activity.

        Returns:
            A tuple of the used and executed activities.
        """
        synapse_activity_used = []
        synapse_activity_executed = []

        for used in self.used:
            if isinstance(used, UsedEntity):
                synapse_activity_used.append(
                    {
                        "reference": {
                            "targetId": used.target_id,
                            "targetVersionNumber": used.target_version_number,
                        }
                    }
                )
            elif isinstance(used, UsedURL):
                synapse_activity_used.append(
                    {
                        "name": used.name,
                        "url": used.url,
                    }
                )

        for executed in self.executed:
            if isinstance(executed, UsedEntity):
                synapse_activity_executed.append(
                    {
                        "reference": {
                            "targetId": executed.target_id,
                            "targetVersionNumber": executed.target_version_number,
                        },
                        "wasExecuted": True,
                    }
                )
            elif isinstance(executed, UsedURL):
                synapse_activity_executed.append(
                    {"name": executed.name, "url": executed.url, "wasExecuted": True}
                )
        return UsedAndExecutedSynapseActivities(
            used=synapse_activity_used, executed=synapse_activity_executed
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Activity_store: {self.name}"
    )
    async def store_async(
        self,
        parent: Optional[Union["Table", "File"]] = None,
        synapse_client: Optional[Synapse] = None,
    ) -> "Activity":
        """
        Store the Activity in Synapse.

        Arguments:
            parent: The parent entity to associate this activity with.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            The activity object.

        Raises:
            ValueError: Raised if both of the following are true:

                - If the parent does not have an ID.
                - If the Activity does not have an ID and ETag.
        """
        # TODO: Input validation: SYNPY-1400
        used_and_executed_activities = (
            self._create_used_and_executed_synapse_activities()
        )

        synapse_activity = Synapse_Activity(
            name=self.name,
            description=self.description,
            used=used_and_executed_activities.used,
            executed=used_and_executed_activities.executed,
        )

        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        if self.id:
            # Despite init in `Synapse_Activity` not accepting an ID/ETAG the
            # `updateActivity` method expects that it exists on the dict
            # and `setProvenance` accepts it as well.
            synapse_activity["id"] = self.id
            synapse_activity["etag"] = self.etag
        if parent:
            saved_activity = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(
                        synapse_client=synapse_client
                    ).setProvenance(
                        entity=parent.id,
                        activity=synapse_activity,
                    ),
                    current_context,
                ),
            )
        else:
            saved_activity = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(
                        synapse_client=synapse_client
                    ).updateActivity(
                        activity=synapse_activity,
                    ),
                    current_context,
                ),
            )
        self.fill_from_dict(synapse_activity=saved_activity)
        Synapse.get_client(synapse_client=synapse_client).logger.debug(
            f"Stored activity {self.id}"
        )

        return self

    @classmethod
    async def from_parent_async(
        cls,
        parent: Union["Table", "File"],
        synapse_client: Optional[Synapse] = None,
    ) -> Union["Activity", None]:
        """
        Get the Activity from Synapse based on the parent entity.

        Arguments:
            parent: The parent entity this activity is associated with. The parent may
                also have a version_number. Gets the most recent version if version is
                omitted.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            The activity object or None if it does not exist.

        Raises:
            ValueError: If the parent does not have an ID.
        """
        # TODO: Input validation: SYNPY-1400
        with tracer.start_as_current_span(name=f"Activity_get: Parent_ID: {parent.id}"):
            loop = asyncio.get_event_loop()
            current_context = context.get_current()
            try:
                synapse_activity = await loop.run_in_executor(
                    None,
                    lambda: run_and_attach_otel_context(
                        lambda: Synapse.get_client(
                            synapse_client=synapse_client
                        ).getProvenance(
                            entity=parent.id,
                            version=parent.version_number,
                        ),
                        current_context,
                    ),
                )
            except SynapseHTTPError as ex:
                if ex.response.status_code == 404:
                    return None
                else:
                    raise ex
            return cls().fill_from_dict(synapse_activity=synapse_activity)

    @classmethod
    async def delete_async(
        cls,
        parent: Union["Table", "File"],
        synapse_client: Optional[Synapse] = None,
    ) -> None:
        """
        Delete the Activity from Synapse. The Activity must be disassociated from
        all entities before it can be deleted. The first step of this delete call
        is to disassociate the Activity from the parent entity. If you have other
        entities that are associated with this Activity you must disassociate them
        by calling this method on them as well. You'll receive an error for all entities
        until the last one which will delete the Activity.

        Arguments:
            parent: The parent entity this activity is associated with.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Raises:
            ValueError: If the parent does not have an ID.
        """
        # TODO: Input validation: SYNPY-1400
        with tracer.start_as_current_span(
            name=f"Activity_delete: Parent_ID: {parent.id}"
        ):
            loop = asyncio.get_event_loop()
            current_context = context.get_current()
            await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(
                        synapse_client=synapse_client
                    ).deleteProvenance(
                        entity=parent.id,
                    ),
                    current_context,
                ),
            )

Functions

from_parent_async(parent, synapse_client=None) async classmethod

Get the Activity from Synapse based on the parent entity.

PARAMETER	DESCRIPTION
parent	The parent entity this activity is associated with. The parent may also have a version_number. Gets the most recent version if version is omitted. TYPE: Union[Table, File]
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Union[Activity, None]	The activity object or None if it does not exist.

RAISES	DESCRIPTION
ValueError	If the parent does not have an ID.

Source code in synapseclient/models/activity.py

@classmethod
async def from_parent_async(
    cls,
    parent: Union["Table", "File"],
    synapse_client: Optional[Synapse] = None,
) -> Union["Activity", None]:
    """
    Get the Activity from Synapse based on the parent entity.

    Arguments:
        parent: The parent entity this activity is associated with. The parent may
            also have a version_number. Gets the most recent version if version is
            omitted.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        The activity object or None if it does not exist.

    Raises:
        ValueError: If the parent does not have an ID.
    """
    # TODO: Input validation: SYNPY-1400
    with tracer.start_as_current_span(name=f"Activity_get: Parent_ID: {parent.id}"):
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        try:
            synapse_activity = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(
                        synapse_client=synapse_client
                    ).getProvenance(
                        entity=parent.id,
                        version=parent.version_number,
                    ),
                    current_context,
                ),
            )
        except SynapseHTTPError as ex:
            if ex.response.status_code == 404:
                return None
            else:
                raise ex
        return cls().fill_from_dict(synapse_activity=synapse_activity)

store_async(parent=None, synapse_client=None) async

Store the Activity in Synapse.

PARAMETER	DESCRIPTION
parent	The parent entity to associate this activity with. TYPE: Optional[Union[Table, File]] DEFAULT: None
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Activity	The activity object.

RAISES	DESCRIPTION
ValueError	Raised if both of the following are true: If the parent does not have an ID. If the Activity does not have an ID and ETag.

Source code in synapseclient/models/activity.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Activity_store: {self.name}"
)
async def store_async(
    self,
    parent: Optional[Union["Table", "File"]] = None,
    synapse_client: Optional[Synapse] = None,
) -> "Activity":
    """
    Store the Activity in Synapse.

    Arguments:
        parent: The parent entity to associate this activity with.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        The activity object.

    Raises:
        ValueError: Raised if both of the following are true:

            - If the parent does not have an ID.
            - If the Activity does not have an ID and ETag.
    """
    # TODO: Input validation: SYNPY-1400
    used_and_executed_activities = (
        self._create_used_and_executed_synapse_activities()
    )

    synapse_activity = Synapse_Activity(
        name=self.name,
        description=self.description,
        used=used_and_executed_activities.used,
        executed=used_and_executed_activities.executed,
    )

    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    if self.id:
        # Despite init in `Synapse_Activity` not accepting an ID/ETAG the
        # `updateActivity` method expects that it exists on the dict
        # and `setProvenance` accepts it as well.
        synapse_activity["id"] = self.id
        synapse_activity["etag"] = self.etag
    if parent:
        saved_activity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).setProvenance(
                    entity=parent.id,
                    activity=synapse_activity,
                ),
                current_context,
            ),
        )
    else:
        saved_activity = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).updateActivity(
                    activity=synapse_activity,
                ),
                current_context,
            ),
        )
    self.fill_from_dict(synapse_activity=saved_activity)
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Stored activity {self.id}"
    )

    return self

delete_async(parent, synapse_client=None) async classmethod

Delete the Activity from Synapse. The Activity must be disassociated from all entities before it can be deleted. The first step of this delete call is to disassociate the Activity from the parent entity. If you have other entities that are associated with this Activity you must disassociate them by calling this method on them as well. You'll receive an error for all entities until the last one which will delete the Activity.

PARAMETER	DESCRIPTION
parent	The parent entity this activity is associated with. TYPE: Union[Table, File]
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RAISES	DESCRIPTION
ValueError	If the parent does not have an ID.

Source code in synapseclient/models/activity.py

@classmethod
async def delete_async(
    cls,
    parent: Union["Table", "File"],
    synapse_client: Optional[Synapse] = None,
) -> None:
    """
    Delete the Activity from Synapse. The Activity must be disassociated from
    all entities before it can be deleted. The first step of this delete call
    is to disassociate the Activity from the parent entity. If you have other
    entities that are associated with this Activity you must disassociate them
    by calling this method on them as well. You'll receive an error for all entities
    until the last one which will delete the Activity.

    Arguments:
        parent: The parent entity this activity is associated with.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Raises:
        ValueError: If the parent does not have an ID.
    """
    # TODO: Input validation: SYNPY-1400
    with tracer.start_as_current_span(
        name=f"Activity_delete: Parent_ID: {parent.id}"
    ):
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).deleteProvenance(
                    entity=parent.id,
                ),
                current_context,
            ),
        )

---

synapseclient.models.Team dataclass

Bases: TeamSynchronousProtocol

Represents a Synapse Team. User definable fields are:

ATTRIBUTE	DESCRIPTION
id	The ID of the team TYPE: Optional[int]
name	The name of the team TYPE: Optional[str]
description	A short description of the team TYPE: Optional[str]
icon	A file handle ID for the icon image of the team TYPE: Optional[str]
can_public_join	True if members can join without an invitation or approval TYPE: Optional[bool]
can_request_membership	True if users can create a membership request to join TYPE: Optional[bool]
etag	Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle concurrent updates Since the E-Tag changes every time an entity is updated it is used to detect when a client's current representation of an entity is out-of-date. TYPE: Optional[str]
created_on	The date this team was created TYPE: Optional[str]
modified_on	The date this team was last modified TYPE: Optional[str]
created_by	The ID of the user that created this team TYPE: Optional[str]
modified_by	The ID of the user that last modified this team TYPE: Optional[str]

Source code in synapseclient/models/team.py

@dataclass
@async_to_sync
class Team(TeamSynchronousProtocol):
    """
    Represents a [Synapse Team](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/Team.html).
    User definable fields are:

    Attributes:
        id: The ID of the team
        name: The name of the team
        description: A short description of the team
        icon: A file handle ID for the icon image of the team
        can_public_join: True if members can join without an invitation or approval
        can_request_membership: True if users can create a membership request to join
        etag: Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
            concurrent updates Since the E-Tag changes every time an entity is updated
            it is used to detect when a client's current representation of an entity
            is out-of-date.
        created_on: The date this team was created
        modified_on: The date this team was last modified
        created_by: The ID of the user that created this team
        modified_by: The ID of the user that last modified this team
    """

    id: Optional[int] = None
    """The ID of the team"""

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

    description: Optional[str] = None
    """A short description of the team"""

    icon: Optional[str] = None
    """A file handle ID for the icon image of the team"""

    can_public_join: Optional[bool] = False
    """True if members can join without an invitation or approval"""

    can_request_membership: Optional[bool] = True
    """True if users can create a membership request to join"""

    etag: Optional[str] = None
    """
    Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
    concurrent updates Since the E-Tag changes every time an entity is updated it is
    used to detect when a client's current representation of an entity is out-of-date.
    """

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

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

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

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

    def fill_from_dict(
        self, synapse_team: Union[Synapse_Team, Dict[str, str]]
    ) -> "Team":
        """
        Converts a response from the REST API into this dataclass.

        Arguments:
            synapse_team: The response from the REST API.

        Returns:
            The Team object.
        """
        self.id = (
            int(synapse_team.get("id", None)) if synapse_team.get("id", None) else None
        )
        self.name = synapse_team.get("name", None)
        self.description = synapse_team.get("description", None)
        self.icon = synapse_team.get("icon", None)
        self.can_public_join = synapse_team.get("canPublicJoin", False)
        self.can_request_membership = synapse_team.get("canRequestMembership", True)
        self.etag = synapse_team.get("etag", None)
        self.created_on = synapse_team.get("createdOn", None)
        self.modified_on = synapse_team.get("modifiedOn", None)
        self.created_by = synapse_team.get("createdBy", None)
        self.modified_by = synapse_team.get("modifiedBy", None)
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Team_Create: {self.name}"
    )
    async def create_async(self, synapse_client: Optional[Synapse] = None) -> "Team":
        """Creates a new team on Synapse.

        Arguments:
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            Team: The Team object.
        """
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        team = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).create_team(
                    name=self.name,
                    description=self.description,
                    icon=self.icon,
                    can_public_join=self.can_public_join,
                    can_request_membership=self.can_request_membership,
                ),
                current_context,
            ),
        )
        self.fill_from_dict(synapse_team=team)
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Team_Delete: {self.id}"
    )
    async def delete_async(self, synapse_client: Optional[Synapse] = None) -> None:
        """Deletes a team from Synapse.

        Arguments:
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            None
        """
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(synapse_client=synapse_client).delete_team(
                    id=self.id,
                ),
                current_context,
            ),
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Team_Get: {self.id if self.id else self.name}"
    )
    async def get_async(self, synapse_client: Optional[Synapse] = None) -> "Team":
        """
        Gets a Team from Synapse by ID or Name. If both are added to the Team instance
        it will use the ID.

        Arguments:
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Raises:
            ValueError: If the Team object has neither an id nor a name.

        Returns:
            Team: The Team object.
        """
        if self.id:
            loop = asyncio.get_event_loop()
            current_context = context.get_current()
            api_team = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(synapse_client=synapse_client).getTeam(
                        id=self.id,
                    ),
                    current_context,
                ),
            )
            return self.fill_from_dict(api_team)
        elif self.name:
            loop = asyncio.get_event_loop()
            current_context = context.get_current()
            api_team = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(synapse_client=synapse_client).getTeam(
                        id=self.name,
                    ),
                    current_context,
                ),
            )
            return self.fill_from_dict(api_team)
        raise ValueError("Team must have either an id or a name")

    @classmethod
    @otel_trace_method(
        method_to_trace_name=lambda cls, id, **kwargs: f"Team_From_Id: {id}"
    )
    async def from_id_async(
        cls, id: int, synapse_client: Optional[Synapse] = None
    ) -> "Team":
        """Gets Team object using its integer id.

        Arguments:
            id: The id of the team.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            Team: The Team object.
        """

        return await cls(id=id).get_async(synapse_client=synapse_client)

    @classmethod
    @otel_trace_method(
        method_to_trace_name=lambda cls, name, **kwargs: f"Team_From_Name: {name}"
    )
    async def from_name_async(
        cls, name: str, synapse_client: Optional[Synapse] = None
    ) -> "Team":
        """Gets Team object using its string name.

        *** You will be unable to retrieve a team by name immediately after its
        creation because the fragment service is eventually consistent. If you need to
        retrieve a team immediately following creation you should use the
        [from_id][synapseclient.models.Team.from_id] method. ***

        Arguments:
            name: The name of the team.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            Team: The Team object.
        """
        return await cls(name=name).get_async(synapse_client=synapse_client)

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Team_Members: {self.name}"
    )
    async def members_async(
        self, synapse_client: Optional[Synapse] = None
    ) -> List[TeamMember]:
        """
        Gets the TeamMembers associated with a team given the ID field on the
        Team instance.

        Arguments:
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            List[TeamMember]: A List of TeamMember objects.
        """
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        team_members = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).getTeamMembers(team=self),
                current_context,
            ),
        )
        team_member_list = [
            TeamMember().fill_from_dict(synapse_team_member=member)
            for member in team_members
        ]
        return team_member_list

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Team_Invite: {self.name}"
    )
    async def invite_async(
        self,
        user: str,
        message: str,
        force: bool = True,
        synapse_client: Optional[Synapse] = None,
    ) -> Dict[str, str]:
        """Invites a user to a team given the ID field on the Team instance.

        Arguments:
            user: The username of the user to invite.
            message: The message to send.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            dict: The invite response.
        """
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        invite = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).invite_to_team(
                    team=self,
                    user=user,
                    message=message,
                    force=force,
                ),
                current_context,
            ),
        )
        return invite

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Team_Open_Invitations: {self.name}"
    )
    async def open_invitations_async(
        self, synapse_client: Optional[Synapse] = None
    ) -> List[Dict[str, str]]:
        """Gets all open invitations for a team given the ID field on the Team instance.

        Arguments:
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            List[dict]: A list of invitations.
        """
        loop = asyncio.get_event_loop()
        current_context = context.get_current()
        invitations = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).get_team_open_invitations(
                    team=self,
                ),
                current_context,
            ),
        )
        return list(invitations)

Functions

create_async(synapse_client=None) async

Creates a new team on Synapse.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Team	The Team object. TYPE: Team

Source code in synapseclient/models/team.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Team_Create: {self.name}"
)
async def create_async(self, synapse_client: Optional[Synapse] = None) -> "Team":
    """Creates a new team on Synapse.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        Team: The Team object.
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    team = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).create_team(
                name=self.name,
                description=self.description,
                icon=self.icon,
                can_public_join=self.can_public_join,
                can_request_membership=self.can_request_membership,
            ),
            current_context,
        ),
    )
    self.fill_from_dict(synapse_team=team)
    return self

delete_async(synapse_client=None) async

Deletes a team from Synapse.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
None	None

Source code in synapseclient/models/team.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Team_Delete: {self.id}"
)
async def delete_async(self, synapse_client: Optional[Synapse] = None) -> None:
    """Deletes a team from Synapse.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        None
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(synapse_client=synapse_client).delete_team(
                id=self.id,
            ),
            current_context,
        ),
    )

from_id_async(id, synapse_client=None) async classmethod

Gets Team object using its integer id.

PARAMETER	DESCRIPTION
id	The id of the team. TYPE: int
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Team	The Team object. TYPE: Team

Source code in synapseclient/models/team.py

@classmethod
@otel_trace_method(
    method_to_trace_name=lambda cls, id, **kwargs: f"Team_From_Id: {id}"
)
async def from_id_async(
    cls, id: int, synapse_client: Optional[Synapse] = None
) -> "Team":
    """Gets Team object using its integer id.

    Arguments:
        id: The id of the team.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        Team: The Team object.
    """

    return await cls(id=id).get_async(synapse_client=synapse_client)

from_name_async(name, synapse_client=None) async classmethod

Gets Team object using its string name.

*** You will be unable to retrieve a team by name immediately after its creation because the fragment service is eventually consistent. If you need to retrieve a team immediately following creation you should use the [from_id][synapseclient.models.Team.from_id] method. ***

PARAMETER	DESCRIPTION
name	The name of the team. TYPE: str
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Team	The Team object. TYPE: Team

Source code in synapseclient/models/team.py

@classmethod
@otel_trace_method(
    method_to_trace_name=lambda cls, name, **kwargs: f"Team_From_Name: {name}"
)
async def from_name_async(
    cls, name: str, synapse_client: Optional[Synapse] = None
) -> "Team":
    """Gets Team object using its string name.

    *** You will be unable to retrieve a team by name immediately after its
    creation because the fragment service is eventually consistent. If you need to
    retrieve a team immediately following creation you should use the
    [from_id][synapseclient.models.Team.from_id] method. ***

    Arguments:
        name: The name of the team.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        Team: The Team object.
    """
    return await cls(name=name).get_async(synapse_client=synapse_client)

members_async(synapse_client=None) async

Gets the TeamMembers associated with a team given the ID field on the Team instance.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[TeamMember]	List[TeamMember]: A List of TeamMember objects.

Source code in synapseclient/models/team.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Team_Members: {self.name}"
)
async def members_async(
    self, synapse_client: Optional[Synapse] = None
) -> List[TeamMember]:
    """
    Gets the TeamMembers associated with a team given the ID field on the
    Team instance.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        List[TeamMember]: A List of TeamMember objects.
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    team_members = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).getTeamMembers(team=self),
            current_context,
        ),
    )
    team_member_list = [
        TeamMember().fill_from_dict(synapse_team_member=member)
        for member in team_members
    ]
    return team_member_list

invite_async(user, message, force=True, synapse_client=None) async

Invites a user to a team given the ID field on the Team instance.

PARAMETER	DESCRIPTION
user	The username of the user to invite. TYPE: str
message	The message to send. TYPE: str
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
dict	The invite response. TYPE: Dict[str, str]

Source code in synapseclient/models/team.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Team_Invite: {self.name}"
)
async def invite_async(
    self,
    user: str,
    message: str,
    force: bool = True,
    synapse_client: Optional[Synapse] = None,
) -> Dict[str, str]:
    """Invites a user to a team given the ID field on the Team instance.

    Arguments:
        user: The username of the user to invite.
        message: The message to send.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        dict: The invite response.
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    invite = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).invite_to_team(
                team=self,
                user=user,
                message=message,
                force=force,
            ),
            current_context,
        ),
    )
    return invite

open_invitations_async(synapse_client=None) async

Gets all open invitations for a team given the ID field on the Team instance.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[Dict[str, str]]	List[dict]: A list of invitations.

Source code in synapseclient/models/team.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Team_Open_Invitations: {self.name}"
)
async def open_invitations_async(
    self, synapse_client: Optional[Synapse] = None
) -> List[Dict[str, str]]:
    """Gets all open invitations for a team given the ID field on the Team instance.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        List[dict]: A list of invitations.
    """
    loop = asyncio.get_event_loop()
    current_context = context.get_current()
    invitations = await loop.run_in_executor(
        None,
        lambda: run_and_attach_otel_context(
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).get_team_open_invitations(
                team=self,
            ),
            current_context,
        ),
    )
    return list(invitations)

---

synapseclient.models.UserProfile dataclass

Bases: UserProfileSynchronousProtocol

UserProfile represents a user's profile in the system.

ATTRIBUTE	DESCRIPTION
id	A foreign key to the ID of the 'principal' object for the user. TYPE: Optional[int]
etag	Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle concurrent updates. Since the E-Tag changes every time an entity is updated it is used to detect when a client's currentrepresentation of an entity is out-of-date. TYPE: Optional[str]
first_name	This person's given name (forename) TYPE: Optional[str]
last_name	This person's family name (surname) TYPE: Optional[str]
emails	The list of user email addresses registered to this user. TYPE: List[str]
open_ids	The list of OpenIds bound to this user's account. TYPE: List[str]
username	A name chosen by the user that uniquely identifies them. TYPE: Optional[str]
display_name	This field is deprecated and will always be null. TYPE: Optional[str]
r_studio_url	URL for RStudio server assigned to the user. TYPE: Optional[str]
summary	A summary description about this person. TYPE: Optional[str]
position	This person's current position title. TYPE: Optional[str]
location	This person's location. TYPE: Optional[str]
industry	The industry/discipline that this person is associated with. TYPE: Optional[str]
company	This person's current affiliation. TYPE: Optional[str]
profile_picure_file_handle_id	The File Handle id of the user's profile picture. TYPE: Optional[str]
url	A link to more information about this person. TYPE: Optional[str]
team_name	This person's default team name. TYPE: Optional[str]
notification_settings	Contains a user's notification settings. TYPE: Optional[str]
preferences	User preferences TYPE: Optional[List[UserPreference]]
created_on	The date this profile was created. TYPE: Optional[str]

Source code in synapseclient/models/user.py

@dataclass()
@async_to_sync
class UserProfile(UserProfileSynchronousProtocol):
    """
    UserProfile represents a user's profile in the system.

    Attributes:
        id: A foreign key to the ID of the 'principal' object for the user.
        etag: Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
            concurrent updates. Since the E-Tag changes every time an entity is updated
            it is used to detect when a client's currentrepresentation of an entity is
            out-of-date.
        first_name: This person's given name (forename)
        last_name: This person's family name (surname)
        emails: The list of user email addresses registered to this user.
        open_ids: The list of OpenIds bound to this user's account.
        username: A name chosen by the user that uniquely identifies them.
        display_name: This field is deprecated and will always be null.
        r_studio_url: URL for RStudio server assigned to the user.
        summary: A summary description about this person.
        position: This person's current position title.
        location: This person's location.
        industry: The industry/discipline that this person is associated with.
        company: This person's current affiliation.
        profile_picure_file_handle_id: The File Handle id of the user's profile picture.
        url: A link to more information about this person.
        team_name: This person's default team name.
        notification_settings: Contains a user's notification settings.
        preferences: User preferences
        created_on: The date this profile was created.
    """

    id: Optional[int] = None
    """A foreign key to the ID of the 'principal' object for the user."""

    username: Optional[str] = None
    """A name chosen by the user that uniquely identifies them."""

    etag: Optional[str] = None
    """
    Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
    concurrent updates. Since the E-Tag changes every time an entity is updated it is
    used to detect when a client's current representation of an entity is out-of-date.
    """

    first_name: Optional[str] = None
    """This person's given name (forename)"""

    last_name: Optional[str] = None
    """This person's family name (surname)"""

    emails: List[str] = field(default_factory=list)
    """The list of user email addresses registered to this user."""

    open_ids: List[str] = field(default_factory=list)
    """The list of OpenIds bound to this user's account."""

    r_studio_url: Optional[str] = None
    """URL for RStudio server assigned to the user"""

    summary: Optional[str] = None
    """A summary description about this person"""

    position: Optional[str] = None
    """This person's current position title"""

    location: Optional[str] = None
    """This person's location"""

    industry: Optional[str] = None
    """The industry/discipline that this person is associated with"""

    company: Optional[str] = None
    """This person's current affiliation"""

    profile_picure_file_handle_id: Optional[str] = None
    """The File Handle id of the user's profile picture"""

    url: Optional[str] = None
    """A link to more information about this person"""

    team_name: Optional[str] = None
    """This person's default team name"""

    send_email_notifications: Optional[bool] = True
    """Should the user receive email notifications? Default true."""

    mark_emailed_messages_as_read: Optional[bool] = False
    """Should messages that are emailed to the user be marked as
    READ in Synapse? Default false."""

    preferences: Optional[List[UserPreference]] = field(default_factory=list)
    """User preferences"""

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

    def fill_from_dict(
        self, synapse_user_profile: Union[Synapse_UserProfile, Dict]
    ) -> "UserProfile":
        """Fills the UserProfile object from a dictionary.

        Arguments:
            synapse_user_profile: The dictionary to fill the UserProfile object from.
                Typically filled from a
                [Synapse UserProfile](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/UserProfile.html) object.
        """
        self.id = (
            int(synapse_user_profile.get("ownerId", None))
            if synapse_user_profile.get("ownerId", None)
            else None
        )
        self.etag = synapse_user_profile.get("etag", None)
        self.first_name = synapse_user_profile.get("firstName", None)
        self.last_name = synapse_user_profile.get("lastName", None)
        self.emails = synapse_user_profile.get("emails", [])
        self.open_ids = synapse_user_profile.get("openIds", [])
        self.username = synapse_user_profile.get("userName", None)
        self.r_studio_url = synapse_user_profile.get("rStudioUrl", None)
        self.summary = synapse_user_profile.get("summary", None)
        self.position = synapse_user_profile.get("position", None)
        self.location = synapse_user_profile.get("location", None)
        self.industry = synapse_user_profile.get("industry", None)
        self.company = synapse_user_profile.get("company", None)
        self.profile_picure_file_handle_id = synapse_user_profile.get(
            "profilePicureFileHandleId", None
        )
        self.url = synapse_user_profile.get("url", None)
        self.team_name = synapse_user_profile.get("teamName", None)
        notification_settings = synapse_user_profile.get("notificationSettings", {})
        self.send_email_notifications = notification_settings.get(
            "sendEmailNotifications", True
        )
        self.mark_emailed_messages_as_read = notification_settings.get(
            "markEmailedMessagesAsRead", False
        )
        synapse_preferences_list = synapse_user_profile.get("preferences", [])
        preferences = []
        for preference in synapse_preferences_list:
            preferences.append(
                UserPreference(
                    name=preference.get("name", None),
                    value=preference.get("value", None),
                )
            )
        self.preferences = preferences
        self.created_on = synapse_user_profile.get("createdOn", None)

        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Profile_Get: Username: {self.username}, id: {self.id}"
    )
    async def get_async(
        self,
        synapse_client: Optional[Synapse] = None,
    ) -> "UserProfile":
        """
        Gets a UserProfile object using its id or username in that order. If an id
        and username is not specified this will retrieve the current user's profile.

        Arguments:
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            The UserProfile object.

        """
        loop = asyncio.get_event_loop()

        current_context = context.get_current()
        if self.id:
            synapse_user_profile = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(
                        synapse_client=synapse_client
                    ).get_user_profile_by_id(id=self.id),
                    current_context,
                ),
            )
        elif self.username:
            synapse_user_profile = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(
                        synapse_client=synapse_client
                    ).get_user_profile_by_username(username=self.username),
                    current_context,
                ),
            )
        else:
            synapse_user_profile = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(
                        synapse_client=synapse_client
                    ).get_user_profile_by_username(),
                    current_context,
                ),
            )

        self.fill_from_dict(synapse_user_profile=synapse_user_profile)
        return self

    @classmethod
    @otel_trace_method(
        method_to_trace_name=lambda cls, user_id, **kwargs: f"Profile_From_Id: {user_id}"
    )
    async def from_id_async(
        cls, user_id: int, synapse_client: Optional[Synapse] = None
    ) -> "UserProfile":
        """Gets UserProfile object using its integer id. Wrapper for the
        [get][synapseclient.models.UserProfile.get] method.

        Arguments:
            user_id: The id of the user.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            The UserProfile object.
        """

        return await cls(id=user_id).get_async(synapse_client=synapse_client)

    @classmethod
    @otel_trace_method(
        method_to_trace_name=lambda cls, username, **kwargs: f"Profile_From_Username: {username}"
    )
    async def from_username_async(
        cls, username: str, synapse_client: Optional[Synapse] = None
    ) -> "UserProfile":
        """
        Gets UserProfile object using its string name. Wrapper for the
        [get][synapseclient.models.UserProfile.get] method.

        Arguments:
            username: A name chosen by the user that uniquely identifies them.
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            The UserProfile object.
        """
        return await cls(username=username).get_async(synapse_client=synapse_client)

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Profile_is_certified: Username: {self.username}, id: {self.id}"
    )
    async def is_certified_async(
        self,
        synapse_client: Optional[Synapse] = None,
    ) -> "bool":
        """
        Determine whether a user is certified.

        Arguments:
            synapse_client: If not passed in or None this will use the last client
                from the `.login()` method.

        Returns:
            True if the user is certified, False otherwise.

        Raises:
            ValueError: If id nor username is specified.
        """
        loop = asyncio.get_event_loop()

        current_context = context.get_current()
        if self.id or self.username:
            is_certified = await loop.run_in_executor(
                None,
                lambda: run_and_attach_otel_context(
                    lambda: Synapse.get_client(
                        synapse_client=synapse_client
                    ).is_certified(user=self.id or self.username),
                    current_context,
                ),
            )
        else:
            raise ValueError("Must specify either id or username")

        return is_certified

Functions

get_async(synapse_client=None) async

Gets a UserProfile object using its id or username in that order. If an id and username is not specified this will retrieve the current user's profile.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
UserProfile	The UserProfile object.

Source code in synapseclient/models/user.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Profile_Get: Username: {self.username}, id: {self.id}"
)
async def get_async(
    self,
    synapse_client: Optional[Synapse] = None,
) -> "UserProfile":
    """
    Gets a UserProfile object using its id or username in that order. If an id
    and username is not specified this will retrieve the current user's profile.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        The UserProfile object.

    """
    loop = asyncio.get_event_loop()

    current_context = context.get_current()
    if self.id:
        synapse_user_profile = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).get_user_profile_by_id(id=self.id),
                current_context,
            ),
        )
    elif self.username:
        synapse_user_profile = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).get_user_profile_by_username(username=self.username),
                current_context,
            ),
        )
    else:
        synapse_user_profile = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).get_user_profile_by_username(),
                current_context,
            ),
        )

    self.fill_from_dict(synapse_user_profile=synapse_user_profile)
    return self

from_id_async(user_id, synapse_client=None) async classmethod

Gets UserProfile object using its integer id. Wrapper for the [get][synapseclient.models.UserProfile.get] method.

PARAMETER	DESCRIPTION
user_id	The id of the user. TYPE: int
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
UserProfile	The UserProfile object.

Source code in synapseclient/models/user.py

@classmethod
@otel_trace_method(
    method_to_trace_name=lambda cls, user_id, **kwargs: f"Profile_From_Id: {user_id}"
)
async def from_id_async(
    cls, user_id: int, synapse_client: Optional[Synapse] = None
) -> "UserProfile":
    """Gets UserProfile object using its integer id. Wrapper for the
    [get][synapseclient.models.UserProfile.get] method.

    Arguments:
        user_id: The id of the user.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        The UserProfile object.
    """

    return await cls(id=user_id).get_async(synapse_client=synapse_client)

from_username_async(username, synapse_client=None) async classmethod

Gets UserProfile object using its string name. Wrapper for the [get][synapseclient.models.UserProfile.get] method.

PARAMETER	DESCRIPTION
username	A name chosen by the user that uniquely identifies them. TYPE: str
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
UserProfile	The UserProfile object.

Source code in synapseclient/models/user.py

@classmethod
@otel_trace_method(
    method_to_trace_name=lambda cls, username, **kwargs: f"Profile_From_Username: {username}"
)
async def from_username_async(
    cls, username: str, synapse_client: Optional[Synapse] = None
) -> "UserProfile":
    """
    Gets UserProfile object using its string name. Wrapper for the
    [get][synapseclient.models.UserProfile.get] method.

    Arguments:
        username: A name chosen by the user that uniquely identifies them.
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        The UserProfile object.
    """
    return await cls(username=username).get_async(synapse_client=synapse_client)

is_certified_async(synapse_client=None) async

Determine whether a user is certified.

PARAMETER	DESCRIPTION
synapse_client	If not passed in or None this will use the last client from the .login() method. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
bool	True if the user is certified, False otherwise.

RAISES	DESCRIPTION
ValueError	If id nor username is specified.

Source code in synapseclient/models/user.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Profile_is_certified: Username: {self.username}, id: {self.id}"
)
async def is_certified_async(
    self,
    synapse_client: Optional[Synapse] = None,
) -> "bool":
    """
    Determine whether a user is certified.

    Arguments:
        synapse_client: If not passed in or None this will use the last client
            from the `.login()` method.

    Returns:
        True if the user is certified, False otherwise.

    Raises:
        ValueError: If id nor username is specified.
    """
    loop = asyncio.get_event_loop()

    current_context = context.get_current()
    if self.id or self.username:
        is_certified = await loop.run_in_executor(
            None,
            lambda: run_and_attach_otel_context(
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).is_certified(user=self.id or self.username),
                current_context,
            ),
        )
    else:
        raise ValueError("Must specify either id or username")

    return is_certified