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: List[File]
folders	Any folders that are at the root directory of the project. TYPE: 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 {} or None and store the entity. 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 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. TYPE: bool
parent_id	The parent ID of the project. In practice projects do not have a parent, but this is required for the inner workings of Synapse. TYPE: Optional[str]

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 `{}` or None and store the entity.
        create_or_update: (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: The parent ID of the project. In practice projects do not have a
            parent, but this is required for the inner workings of Synapse.

    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: List["File"] = field(default_factory=list, compare=False)
    """Any files that are at the root directory of the project."""

    folders: 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_factory=dict, 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 `{}`
    or None and store the entity."""

    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 {}
        )

    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", {})
            )
        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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)
        trace.get_current_span().set_attributes(
            {
                "synapse.name": self.name or "",
                "synapse.id": self.id or "",
            }
        )
        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)
            entity = await loop.run_in_executor(
                None,
                lambda: Synapse.get_client(synapse_client=synapse_client).store(
                    obj=synapse_project,
                    set_annotations=False,
                    createOrUpdate=False,
                ),
            )
            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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

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

        await get_from_entity_factory(
            entity_to_update=self,
            synapse_id_or_path=entity_id,
            synapse_client=synapse_client,
        )

        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            None

        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()
        await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).delete(
                obj=entity_id,
            ),
        )

    @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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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()

        syn = Synapse.get_client(synapse_client=synapse_client)
        source_and_destination = await loop.run_in_executor(
            None,
            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,
            ),
        )

        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(synapse_client=syn)
        ).sync_from_synapse_async(
            download_file=False,
            synapse_client=synapse_client,
        )
        syn.logger.debug(f"Copied from project {self.id} to {destination_id}")
        return project_copy

Functions

get_async async

get_async(*, synapse_client: Optional[Synapse] = None) -> Project

Get the project metadata from Synapse.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

    await get_from_entity_factory(
        entity_to_update=self,
        synapse_id_or_path=entity_id,
        synapse_client=synapse_client,
    )

    self._set_last_persistent_instance()
    return self

store_async async

store_async(failure_strategy: 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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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)
    trace.get_current_span().set_attributes(
        {
            "synapse.name": self.name or "",
            "synapse.id": self.id or "",
        }
    )
    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)
        entity = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).store(
                obj=synapse_project,
                set_annotations=False,
                createOrUpdate=False,
            ),
        )
        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 async

delete_async(*, synapse_client: Optional[Synapse] = None) -> None

Delete the project from Synapse.

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

RETURNS	DESCRIPTION
None	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        None

    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()
    await loop.run_in_executor(
        None,
        lambda: Synapse.get_client(synapse_client=synapse_client).delete(
            obj=entity_id,
        ),
    )

sync_from_synapse_async async

sync_from_synapse_async(path: Optional[str] = None, recursive: bool = True, download_file: bool = True, if_collision: str = COLLISION_OVERWRITE_LOCAL, failure_strategy: FailureStrategy = LOG_EXCEPTION, include_activity: bool = True, follow_link: bool = False, link_hops: int = 1, queue: Queue = None, *, 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.

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
include_activity	Whether to include the activity of the files. TYPE: bool DEFAULT: True
follow_link	Whether to follow a link entity or not. Links can be used to point at other Synapse entities. TYPE: bool DEFAULT: False
link_hops	The number of hops to follow the link. A number of 1 is used to prevent circular references. There is nothing in place to prevent infinite loops. Be careful if setting this above 1. TYPE: int DEFAULT: 1
queue	An optional queue to use to download files in parallel. TYPE: Queue DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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
            else Child is Link and hops > 0
                note over sync_from_synapse: Append task to follow link
            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()`
            and `_follow_link`
                sync_from_synapse ->>client: call `get_entity_id_bundle2` function
                client-->sync_from_synapse: .
                note over sync_from_synapse: Do nothing if not link
                note over sync_from_synapse: call `_create_task_for_child` and execute
            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,
    include_activity: bool = True,
    follow_link: bool = False,
    link_hops: int = 1,
    queue: asyncio.Queue = None,
    *,
    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.
        include_activity: Whether to include the activity of the files.
        follow_link: Whether to follow a link entity or not. Links can be used to
            point at other Synapse entities.
        link_hops: The number of hops to follow the link. A number of 1 is used to
            prevent circular references. There is nothing in place to prevent
            infinite loops. Be careful if setting this above 1.
        queue: An optional queue to use to download files in parallel.
        synapse_client: If not passed in and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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
                else Child is Link and hops > 0
                    note over sync_from_synapse: Append task to follow link
                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()`
                and `_follow_link`
                    sync_from_synapse ->>client: call `get_entity_id_bundle2` function
                    client-->sync_from_synapse: .
                    note over sync_from_synapse: Do nothing if not link
                    note over sync_from_synapse: call `_create_task_for_child` and execute
                end
            end

        deactivate sync_from_synapse
        deactivate project_or_folder
    ```

    """
    syn = Synapse.get_client(synapse_client=synapse_client)
    custom_message = "Syncing from Synapse" if not download_file else None
    with shared_download_progress_bar(
        file_size=1, synapse_client=syn, custom_message=custom_message
    ):
        return await self._sync_from_synapse_async(
            path=path,
            recursive=recursive,
            download_file=download_file,
            if_collision=if_collision,
            failure_strategy=failure_strategy,
            include_activity=include_activity,
            follow_link=follow_link,
            link_hops=link_hops,
            queue=queue,
            synapse_client=syn,
        )

get_permissions_async async

get_permissions_async(*, synapse_client: Optional[Synapse] = None) -> Permissions

Get the permissions that the caller has on an Entity.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        A 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()

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

get_acl_async async

get_acl_async(principal_id: int = None, *, synapse_client: Optional[Synapse] = None) -> List[str]

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

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

set_permissions_async async

set_permissions_async(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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, list]]	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        An 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()

    return await loop.run_in_executor(
        None,
        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,
        ),
    )

---

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: List[File]
folders	Folders that exist within this folder. TYPE: 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 {} or None and store the entity. 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 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. TYPE: bool

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 `{}` or None and store the entity.
        create_or_update: (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.
    """

    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: List["File"] = field(default_factory=list, compare=False)
    """Files that exist within this folder."""

    folders: 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_factory=dict, 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 `{}`
    or None and store the entity."""

    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 {}
        )

    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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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(
                    synapse_client=synapse_client
                )
            )
        ):
            merge_dataclass_entities(source=existing_folder, destination=self)
        trace.get_current_span().set_attributes(
            {
                "synapse.name": self.name or "",
                "synapse.id": self.id or "",
            }
        )
        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)
            entity = await loop.run_in_executor(
                None,
                lambda: Synapse.get_client(synapse_client=synapse_client).store(
                    obj=synapse_folder,
                    set_annotations=False,
                    isRestricted=self.is_restricted,
                    createOrUpdate=False,
                ),
            )

            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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

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

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

        await get_from_entity_factory(
            entity_to_update=self,
            synapse_id_or_path=entity_id,
            synapse_client=synapse_client,
        )

        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            None

        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()
        await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).delete(
                obj=self.id,
            ),
        )

    @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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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()

        syn = Synapse.get_client(synapse_client=synapse_client)
        source_and_destination = await loop.run_in_executor(
            None,
            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,
            ),
        )

        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(synapse_client=synapse_client)
        ).sync_from_synapse_async(
            download_file=False,
            synapse_client=synapse_client,
        )
        syn.logger.debug(
            f"Copied from folder {self.id} to {parent_id} with new id of {folder_copy.id}"
        )
        return folder_copy

Functions

get_async async

get_async(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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

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

    await get_from_entity_factory(
        entity_to_update=self,
        synapse_id_or_path=entity_id,
        synapse_client=synapse_client,
    )

    self._set_last_persistent_instance()
    return self

store_async async

store_async(parent: Optional[Union[Folder, Project]] = None, failure_strategy: 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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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(
                synapse_client=synapse_client
            )
        )
    ):
        merge_dataclass_entities(source=existing_folder, destination=self)
    trace.get_current_span().set_attributes(
        {
            "synapse.name": self.name or "",
            "synapse.id": self.id or "",
        }
    )
    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)
        entity = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).store(
                obj=synapse_folder,
                set_annotations=False,
                isRestricted=self.is_restricted,
                createOrUpdate=False,
            ),
        )

        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 async

delete_async(*, synapse_client: Optional[Synapse] = None) -> None

Delete the folder from Synapse by its id.

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

RETURNS	DESCRIPTION
None	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        None

    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()
    await loop.run_in_executor(
        None,
        lambda: Synapse.get_client(synapse_client=synapse_client).delete(
            obj=self.id,
        ),
    )

copy_async async

copy_async(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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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()

    syn = Synapse.get_client(synapse_client=synapse_client)
    source_and_destination = await loop.run_in_executor(
        None,
        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,
        ),
    )

    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(synapse_client=synapse_client)
    ).sync_from_synapse_async(
        download_file=False,
        synapse_client=synapse_client,
    )
    syn.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 async

sync_from_synapse_async(path: Optional[str] = None, recursive: bool = True, download_file: bool = True, if_collision: str = COLLISION_OVERWRITE_LOCAL, failure_strategy: FailureStrategy = LOG_EXCEPTION, include_activity: bool = True, follow_link: bool = False, link_hops: int = 1, queue: Queue = None, *, 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.

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
include_activity	Whether to include the activity of the files. TYPE: bool DEFAULT: True
follow_link	Whether to follow a link entity or not. Links can be used to point at other Synapse entities. TYPE: bool DEFAULT: False
link_hops	The number of hops to follow the link. A number of 1 is used to prevent circular references. There is nothing in place to prevent infinite loops. Be careful if setting this above 1. TYPE: int DEFAULT: 1
queue	An optional queue to use to download files in parallel. TYPE: Queue DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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
            else Child is Link and hops > 0
                note over sync_from_synapse: Append task to follow link
            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()`
            and `_follow_link`
                sync_from_synapse ->>client: call `get_entity_id_bundle2` function
                client-->sync_from_synapse: .
                note over sync_from_synapse: Do nothing if not link
                note over sync_from_synapse: call `_create_task_for_child` and execute
            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,
    include_activity: bool = True,
    follow_link: bool = False,
    link_hops: int = 1,
    queue: asyncio.Queue = None,
    *,
    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.
        include_activity: Whether to include the activity of the files.
        follow_link: Whether to follow a link entity or not. Links can be used to
            point at other Synapse entities.
        link_hops: The number of hops to follow the link. A number of 1 is used to
            prevent circular references. There is nothing in place to prevent
            infinite loops. Be careful if setting this above 1.
        queue: An optional queue to use to download files in parallel.
        synapse_client: If not passed in and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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
                else Child is Link and hops > 0
                    note over sync_from_synapse: Append task to follow link
                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()`
                and `_follow_link`
                    sync_from_synapse ->>client: call `get_entity_id_bundle2` function
                    client-->sync_from_synapse: .
                    note over sync_from_synapse: Do nothing if not link
                    note over sync_from_synapse: call `_create_task_for_child` and execute
                end
            end

        deactivate sync_from_synapse
        deactivate project_or_folder
    ```

    """
    syn = Synapse.get_client(synapse_client=synapse_client)
    custom_message = "Syncing from Synapse" if not download_file else None
    with shared_download_progress_bar(
        file_size=1, synapse_client=syn, custom_message=custom_message
    ):
        return await self._sync_from_synapse_async(
            path=path,
            recursive=recursive,
            download_file=download_file,
            if_collision=if_collision,
            failure_strategy=failure_strategy,
            include_activity=include_activity,
            follow_link=follow_link,
            link_hops=link_hops,
            queue=queue,
            synapse_client=syn,
        )

get_permissions_async async

get_permissions_async(*, synapse_client: Optional[Synapse] = None) -> Permissions

Get the permissions that the caller has on an Entity.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        A 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()

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

get_acl_async async

get_acl_async(principal_id: int = None, *, synapse_client: Optional[Synapse] = None) -> List[str]

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

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

set_permissions_async async

set_permissions_async(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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, list]]	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        An 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()

    return await loop.run_in_executor(
        None,
        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,
        ),
    )

---

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. Using shorthand ~ will be expanded to the user's home directory. This is used during a get operation to specify where to download the file to. It should be pointing to a directory. This is also used during a store operation to specify the file to upload. It should be pointing to a file. TYPE: Optional[str]
description	The description of this file. Must be 1000 characters or less. 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_label	The version label for this entity. Updates to the entity will increment the version number. TYPE: Optional[str]
version_comment	The version comment for this entity TYPE: Optional[str]
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]
external_url	The external URL of this file. If this is set AND synapse_store is False, only a reference to this URL and the file metadata will be stored in Synapse. The file itself will not be uploaded. If this attribute is set it will override the path. TYPE: Optional[str]
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. Activity cannot be removed during a store operation by setting it to None. You must use: synapseclient.models.Activity.delete_async or [synapseclient.models.Activity.disassociate_from_entity_async][]. 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 {} or None and store the entity. TYPE: Optional[Dict[str, Union[List[str], List[bool], List[float], List[int], List[date], List[datetime]]]]

ATTRIBUTE	DESCRIPTION
content_type	(New Upload Only) 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. This can be specified only during the initial store of this file or any time there is a new file to upload. In order to change this after the File has been created use synapseclient.models.File.change_metadata. TYPE: Optional[str]
content_size	(New Upload Only) The size of the file in bytes. This can be specified only during the initial creation of the File. This is also only applicable to files not uploaded to Synapse. ie: synapse_store is False. TYPE: Optional[int]

ATTRIBUTE	DESCRIPTION
content_md5	(Store only) The MD5 of the file is known. If not supplied this will be computed in the client is possible. If supplied for a file entity already stored in Synapse it will be calculated again to check if a new upload needs to occur. This will not be filled in during a read for data. It is only used during a store operation. To retrieve the md5 of the file after read from synapse use the .file_handle.content_md5 attribute. TYPE: Optional[str]
create_or_update	(Store only) Indicates whether the method should automatically perform an update if the file conflicts with an existing Synapse object. TYPE: bool
force_version	(Store only) Indicates whether the method should increment the version of the object if something within the entity has changed. For example updating the description or name. You may set this to False and an update to the entity will not increment the version. Updating the version_label attribute will also cause a version update regardless of this flag. 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. This may be used only by an administrator of the specified file. TYPE: bool
merge_existing_annotations	(Store only) Works in conjunction with create_or_update in that this is only evaluated if create_or_update is True. If this entity exists in Synapse that has annotations that are not present in a store operation, these annotations will be added to the entity. If this is False any annotations that are not present within a store operation will be removed from this entity. This allows one to complete a destructive update of annotations on an entity. TYPE: bool
associate_activity_to_new_version	(Store only) Works in conjunction with create_or_update in that this is only evaluated if create_or_update is True. When true an activity already attached to the current version of this entity will be associated the new version during a store operation if the version was updated. This is useful if you are updating the entity and want to ensure that the activity is persisted onto the new version the entity. 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

ATTRIBUTE	DESCRIPTION
download_file	(Get only) If True the file will be downloaded. TYPE: bool
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]

ATTRIBUTE	DESCRIPTION
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]
version_number	(Read Only) The version number issued to this version on the object. TYPE: Optional[int]
is_latest_version	(Read Only) If this is the latest version of the object. TYPE: Optional[bool]
file_handle	(Read Only) The file handle associated with this entity. TYPE: Optional[FileHandle]

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. Using shorthand `~` will be expanded to the
            user's home directory.

            This is used during a `get` operation to specify where to download the file
            to. It should be pointing to a directory.

            This is also used during a `store` operation to specify the file to upload.
            It should be pointing to a file.

        description: The description of this file. Must be 1000 characters or less.
        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_label: The version label for this entity. Updates to the entity will
            increment the version number.
        version_comment: The version comment for this entity
        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.
        external_url: The external URL of this file. If this is set AND `synapse_store`
            is False, only a reference to this URL and the file metadata will be stored
            in Synapse. The file itself will not be uploaded. If this attribute is set
            it will override the `path`.
        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. Activity
            cannot be removed during a store operation by setting it to None. You must
            use: [synapseclient.models.Activity.delete_async][] or
            [synapseclient.models.Activity.disassociate_from_entity_async][].
        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 `{}` or None and store the entity.

    Attributes:
        content_type: (New Upload Only)
            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.


            This can be specified only during the initial store of this file or any time
            there is a new file to upload.
            In order to change this after the File has been created use
            [synapseclient.models.File.change_metadata][].
        content_size: (New Upload Only)
            The size of the file in bytes. This can be specified only during the initial
            creation of the File. This is also only applicable to files not uploaded to
            Synapse. ie: `synapse_store` is False.

    Attributes:
        content_md5: (Store only) The MD5 of the file is known. If not supplied this
            will be computed in the client is possible. If supplied for a file entity
            already stored in Synapse it will be calculated again to check if a new
            upload needs to occur. This will not be filled in during a read for data. It
            is only used during a store operation. To retrieve the md5 of the file after
            read from synapse use the `.file_handle.content_md5` attribute.
        create_or_update: (Store only)
            Indicates whether the method should automatically perform an
            update if the `file` conflicts with an existing Synapse object.
        force_version: (Store only)
            Indicates whether the method should increment the version of the object if
            something within the entity has changed. For example updating the
            description or name. You may set this to False and an update to the
            entity will not increment the version.

            Updating the `version_label` attribute will also cause a version update
            regardless  of this flag.

            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.

            This may be used only by an administrator of the specified file.
        merge_existing_annotations: (Store only)
            Works in conjunction with `create_or_update` in that this is only evaluated
            if `create_or_update` is True. If this entity exists in Synapse that has
            annotations that are not present in a store operation, these annotations
            will be added to the entity. If this is False any annotations that are not
            present within a store operation will be removed from this entity. This
            allows one to complete a destructive update of annotations on an entity.
        associate_activity_to_new_version: (Store only)
            Works in conjunction with `create_or_update` in that this is only evaluated
            if `create_or_update` is True. When true an activity already attached to the
            current version of this entity will be associated the new version during a
            store operation if the version was updated. This is useful if you are
            updating the entity and want to ensure that the activity is persisted onto
            the new version the entity.
        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.

    Attributes:
        download_file: (Get only) If True the file will be downloaded.
        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.

    Attributes:
        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.
        version_number: (Read Only) The version number issued to this version on the
            object.
        is_latest_version: (Read Only) If this is the latest version of the object.
        file_handle: (Read Only) The file handle associated with this entity.
    """

    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] = field(default=None, compare=False)
    """The path to the file on disk. Using shorthand `~` will be expanded to the user's
    home directory.

    This is used during a `get` operation to specify where to download the file to. It
    should be pointing to a directory.

    This is also used during a `store` operation to specify the file to upload. It
    should be pointing to a file."""

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

    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_label: Optional[str] = None
    """The version label for this entity. Updates to the entity will increment the
    version number."""

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

    data_file_handle_id: Optional[str] = None
    """
    ID of the file handle 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.
    """

    external_url: Optional[str] = field(default=None, compare=False)
    """
    The external URL of this file. If this is set AND `synapse_store` is False, only
    a reference to this URL and the file metadata will be stored in Synapse. The file
    itself will not be uploaded. If this attribute is set it will override the `path`.
    """

    activity: Optional[Activity] = field(default=None, compare=False)
    """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. Activity cannot
    be removed during a store operation by setting it to None. You must use:
    [synapseclient.models.Activity.delete_async][] or
    [synapseclient.models.Activity.disassociate_from_entity_async][].
    """

    annotations: Optional[
        Dict[
            str,
            Union[
                List[str],
                List[bool],
                List[float],
                List[int],
                List[date],
                List[datetime],
            ],
        ]
    ] = field(default_factory=dict, 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 `{}`."""

    content_type: Optional[str] = None
    """
    (New Upload Only)
    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.

    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][].
    """

    content_size: Optional[int] = None
    """
    (New Upload Only)
    The size of the file in bytes. This can be specified only during the initial
    creation of the File. This is also only applicable to files not uploaded to Synapse.
    ie: `synapse_store` is False.
    """

    content_md5: Optional[str] = field(default=None, compare=False)
    """
    (Store only)
    The MD5 of the file is known. If not supplied this will be computed in the client
    is possible. If supplied for a file entity already stored in Synapse it will be
    calculated again to check if a new upload needs to occur. This will not be filled
    in during a read for data. It is only used during a store operation. To retrieve
    the md5 of the file after read from synapse use the `.file_handle.content_md5`
    attribute.
    """

    create_or_update: bool = field(default=True, repr=False, compare=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, compare=False)
    """
    (Store only)

    Indicates whether the method should increment the version of the object if something
    within the entity has changed. For example updating the description or name.
    You may set this to False and an update to the entity will not increment the
    version.

    Updating the `version_label` attribute will also cause a version update regardless
    of this flag.

    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.

    This may be used only by an administrator of the specified file.
    """

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

    Works in conjunction with `create_or_update` in that this is only evaluated if
    `create_or_update` is True. If this entity exists in Synapse that has annotations
    that are not present in a store operation, these annotations will be added to the
    entity. If this is False any annotations that are not present within a store
    operation will be removed from this entity. This allows one to complete a
    destructive update of annotations on an entity.
    """

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

    Works in conjunction with `create_or_update` in that this is only evaluated if
    `create_or_update` is True. When true an activity already attached to the current
    version of this entity will be associated the new version during a store operation
    if the version was updated. This is useful if you are updating the entity and want
    to ensure that the activity is persisted onto the new version the entity.

    When this is False the activity will not be associated to the new version of the
    entity during a store operation.

    Regardless of this setting, if you have an Activity object on the entity it will be
    persisted onto the new version. This is only used when you don't have an Activity
    object on the entity.
    """

    _present_manifest_fields: List[str] = field(default=None, repr=False, compare=False)
    """Hidden attribute to pass along what columns were present in a manifest upload."""

    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, compare=False)
    """
    (Get only)

    If True the file will be downloaded."""

    if_collision: str = field(default="keep.both", repr=False, compare=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, compare=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."""

    etag: Optional[str] = field(default=None, compare=False)
    """
    (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] = 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."""

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

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

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

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

    @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 = 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 {}
        )

    def _fill_from_file_handle(self) -> None:
        """Fill the file object from the file handle."""
        if self.file_handle:
            self.data_file_handle_id = self.file_handle.id
            self.content_type = self.file_handle.content_type
            self.content_size = self.file_handle.content_size
            self.external_url = self.file_handle.external_url

    def fill_from_dict(
        self,
        synapse_file: Union[Synapse_File, Dict[str, Union[bool, str, int]]],
        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.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)
        self.path = synapse_file.get("path", self.path)
        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.fill_from_dict(
                synapse_instance=synapse_file_handle
            )
            self._fill_from_file_handle()

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

    def _cannot_store(self) -> bool:
        """Determines based on some guard conditions if we are unable to continue with
        a store operation."""
        return (
            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 self.parent_id is not None)
            and not (
                self.parent_id is not None and self.data_file_handle_id is not None
            )
        )

    async def _load_local_md5(self) -> None:
        """Load the MD5 of the file if it's a local file and we have not already loaded
        it."""
        if not self.content_md5 and self.path and os.path.isfile(self.path):
            self.content_md5 = utils.md5_for_file_hex(filename=self.path)

    async def _find_existing_file(
        self, *, synapse_client: Optional[Synapse] = None
    ) -> Union["File", None]:
        """Determines if the file already exists in Synapse. If it does it will return
        the file object, otherwise it will return None. This is used to determine if the
        file should be updated or created."""

        async def get_file(existing_id: str) -> "File":
            """Small wrapper to retrieve a file instance without raising an error if it
            does not exist.

            Arguments:
                existing_id: The ID of the file to retrieve.

            Returns:
                The file object if it exists, otherwise None.
            """
            try:
                file_copy = File(
                    id=existing_id,
                    download_file=False,
                    version_number=self.version_number,
                    synapse_container_limit=self.synapse_container_limit,
                    parent_id=self.parent_id,
                )
                return await file_copy.get_async(
                    synapse_client=synapse_client,
                    include_activity=self.activity is not None
                    or self.associate_activity_to_new_version,
                )
            except SynapseFileNotFoundError:
                return None

        if (
            self.create_or_update
            and not self._last_persistent_instance
            and (
                existing_file_id := await get_id(
                    entity=self,
                    failure_strategy=None,
                    synapse_client=synapse_client,
                )
            )
            and (existing_file := await get_file(existing_file_id))
        ):
            return existing_file
        return None

    def _determine_fields_to_ignore_in_merge(self) -> List[str]:
        """This is used to determine what fields should not be merged when merging two
        entities. This allows for a fine tuned destructive update of an entity.

        This also has special handling during a manifest upload of files. If a manifest
        is specifying fields we'll use those values rather than copying them from the
        existing entity. This is to allow for a destructive update of an entity.

        """
        fields_to_not_merge = []
        if not self.merge_existing_annotations:
            fields_to_not_merge.append("annotations")

        if not self.associate_activity_to_new_version:
            fields_to_not_merge.append("activity")

        if self._present_manifest_fields:
            if "name" in self._present_manifest_fields:
                fields_to_not_merge.append("name")

            if "contentType" in self._present_manifest_fields:
                fields_to_not_merge.append("content_type")

        return fields_to_not_merge

    @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.

        If no Name is specified this will be derived from the file name. This is the
        reccommended way to store a file in Synapse.

        Please note:
        The file, as it appears on disk, will be the file that is downloaded from
        Synapse. The name of the actual File is different from the name of the File
        Entity in Synapse. It is generally not reccommended to specify a different
        name for the Entity and the file as it will cause confusion and potential
        conflicts later on.

        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The file object.

        Raises:
            ValueError: If the file does not have an ID and a path, or a path and a
                parent ID, or a data file handle ID and a parent ID.

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

            File with a parent and existing file handle (This allows multiple entities to reference the underlying file):

                file_instance = await File(data_file_handle_id="123", parent_id="syn456").store_async()

            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). Is is reccommended that `name` and
                `download_as` match to prevent confusion later on:

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

        """
        self.parent_id = parent.id if parent else self.parent_id
        if self._cannot_store():
            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)), or a "
                "(data_file_handle_id with a (`parent_id` or parent with an id)) to store."
            )
        self.name = self.name or (guess_file_name(self.path) if self.path else None)
        client = Synapse.get_client(synapse_client=synapse_client)

        if existing_file := await self._find_existing_file(synapse_client=client):
            merge_dataclass_entities(
                source=existing_file,
                destination=self,
                fields_to_ignore=self._determine_fields_to_ignore_in_merge(),
            )

        if self.path:
            self.path = os.path.expanduser(self.path)
            async with client._get_parallel_file_transfer_semaphore(
                asyncio_event_loop=asyncio.get_running_loop()
            ):
                await self._upload_file(synapse_client=client)
        elif self.data_file_handle_id:
            self.path = client.cache.get(file_handle_id=self.data_file_handle_id)

        if self.has_changed:
            synapse_file = Synapse_File(
                id=self.id,
                path=self.path,
                description=self.description,
                etag=self.etag,
                name=self.name,
                parent=parent.id if parent else self.parent_id,
                contentType=self.content_type,
                contentSize=self.content_size,
                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)

            entity = await store_entity(
                resource=self, entity=synapse_file, synapse_client=client
            )

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

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

        self._set_last_persistent_instance()

        client.logger.debug(f"Stored File {self.name}, id: {self.id}: {self.path}")
        # Clear the content_md5 so that it is recalculated if the file is updated
        self.content_md5 = None
        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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.")
        from synapseutils.copy_functions import changeFileMetaData

        loop = asyncio.get_event_loop()

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

        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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 want to download a file to this directory: "path/to/directory":

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

        await self._load_local_md5()

        await get_from_entity_factory(
            entity_to_update=self,
            synapse_id_or_path=self.id or self.path,
            version=self.version_number,
            if_collision=self.if_collision,
            limit_search=self.synapse_container_limit or self.parent_id,
            download_file=self.download_file,
            download_location=os.path.dirname(self.path)
            if self.path and os.path.isfile(self.path)
            else self.path,
            md5=self.content_md5,
            synapse_client=syn,
        )

        if (
            self.data_file_handle_id
            and (not self.path or (self.path and not os.path.isfile(self.path)))
            and (cached_path := syn.cache.get(file_handle_id=self.data_file_handle_id))
        ):
            self.path = cached_path

        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            None

        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()
        await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).delete(
                obj=self.id,
                version=self.version_number if version_only else None,
            ),
        )
        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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.")
        from synapseutils.copy_functions import copy

        loop = asyncio.get_event_loop()

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

        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

    async def _needs_upload(self, syn: Synapse) -> bool:
        """
        Determines if a file needs to be uploaded to Synapse. The following conditions
        apply:

        - The file exists and is an ExternalFileHandle and the url has changed
        - The file exists and is a local file and the MD5 has changed
        - The file is not present in Synapse

        If the file is already specifying a data_file_handle_id then it is assumed that
        the file is already uploaded to Synapse. It does not need to be uploaded and
        the only thing that will occur is the File metadata will be added to Synapse
        outside of this upload process.

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

        Returns:
            True if the file needs to be uploaded, otherwise False.
        """
        needs_upload = False
        # Check if the file should be uploaded
        if self._last_persistent_instance is not None:
            if (
                self.file_handle
                and self.file_handle.concrete_type
                == "org.sagebionetworks.repo.model.file.ExternalFileHandle"
            ):
                # switching away from ExternalFileHandle or the url was updated
                needs_upload = self.synapse_store or (
                    self.file_handle.external_url != self.external_url
                )
            else:
                # Check if we need to upload a new version of an existing
                # file. If the file referred to by entity['path'] has been
                # modified, we want to upload the new version.
                # If synapeStore is false then we must upload a ExternalFileHandle
                needs_upload = (
                    not self.synapse_store
                    or not self.file_handle
                    or not (
                        exists_in_cache := syn.cache.contains(
                            self.file_handle.id, self.path
                        )
                    )
                )

                md5_stored_in_synapse = (
                    self.file_handle.content_md5 if self.file_handle else None
                )

                # Check if we got an MD5 checksum from Synapse and compare it to the local file
                if (
                    self.synapse_store
                    and needs_upload
                    and os.path.isfile(self.path)
                    and md5_stored_in_synapse
                ):
                    await self._load_local_md5()
                    if md5_stored_in_synapse == (
                        local_file_md5_hex := self.content_md5
                    ):
                        needs_upload = False

                    # If we had a cache miss, but already uploaded to Synapse we
                    # can add the file to the cache.
                    if (
                        not exists_in_cache
                        and self.file_handle
                        and self.file_handle.id
                        and local_file_md5_hex
                    ):
                        syn.cache.add(
                            file_handle_id=self.file_handle.id,
                            path=self.path,
                            md5=local_file_md5_hex,
                        )
        elif self.data_file_handle_id is not None:
            needs_upload = False
        else:
            needs_upload = True
        return needs_upload

    async def _upload_file(
        self,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> "File":
        """The upload process for a file. This will upload the file to Synapse if it
        needs to be uploaded. If the file does not need to be uploaded the file
        metadata will be added to Synapse outside of this upload process.

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

        Returns:
            The file object.
        """
        syn = Synapse.get_client(synapse_client=synapse_client)

        needs_upload = await self._needs_upload(syn=syn)

        if needs_upload:
            parent_id_for_upload = self.parent_id

            if not parent_id_for_upload:
                raise SynapseMalformedEntityError(
                    "Entities of type File must have a parentId."
                )

            updated_file_handle = await upload_file_handle(
                syn=syn,
                parent_entity_id=parent_id_for_upload,
                path=(
                    self.path
                    if (self.synapse_store or self.external_url is None)
                    else self.external_url
                ),
                synapse_store=self.synapse_store,
                md5=self.content_md5,
                file_size=self.content_size,
                mimetype=self.content_type,
            )

            self.file_handle = FileHandle().fill_from_dict(updated_file_handle)
            self._fill_from_file_handle()

        return self

    def _convert_into_legacy_file(self) -> SynapseFile:
        """Convert the file object into a SynapseFile object."""
        return_data = SynapseFile(
            id=self.id,
            name=self.name,
            description=self.description,
            etag=self.etag,
            createdOn=self.created_on,
            modifiedOn=self.modified_on,
            createdBy=self.created_by,
            modifiedBy=self.modified_by,
            parentId=self.parent_id,
            versionNumber=self.version_number,
            versionLabel=self.version_label,
            versionComment=self.version_comment,
            dataFileHandleId=self.data_file_handle_id,
            path=self.path,
            properties={
                "isLatestVersion": self.is_latest_version,
            },
            _file_handle=(
                self.file_handle._convert_into_legacy_file_handle()
                if self.file_handle
                else None
            ),
            annotations=self.annotations,
        )
        delete_none_keys(return_data)
        return return_data

Functions

get_async async

get_async(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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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 want to download a file to this directory: "path/to/directory":

file_instance = await File(path="path/to/directory").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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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 want to download a file to this directory: "path/to/directory":

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

    await self._load_local_md5()

    await get_from_entity_factory(
        entity_to_update=self,
        synapse_id_or_path=self.id or self.path,
        version=self.version_number,
        if_collision=self.if_collision,
        limit_search=self.synapse_container_limit or self.parent_id,
        download_file=self.download_file,
        download_location=os.path.dirname(self.path)
        if self.path and os.path.isfile(self.path)
        else self.path,
        md5=self.content_md5,
        synapse_client=syn,
    )

    if (
        self.data_file_handle_id
        and (not self.path or (self.path and not os.path.isfile(self.path)))
        and (cached_path := syn.cache.get(file_handle_id=self.data_file_handle_id))
    ):
        self.path = cached_path

    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 async

store_async(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.

If no Name is specified this will be derived from the file name. This is the reccommended way to store a file in Synapse.

Please note: The file, as it appears on disk, will be the file that is downloaded from Synapse. The name of the actual File is different from the name of the File Entity in Synapse. It is generally not reccommended to specify a different name for the Entity and the file as it will cause confusion and potential conflicts later on.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
File	The file object.

RAISES	DESCRIPTION
ValueError	If the file does not have an ID and a path, or a path and a parent ID, or a data file handle ID and a parent ID.

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

File with a parent and existing file handle (This allows multiple entities to reference the underlying file):

file_instance = await File(data_file_handle_id="123", parent_id="syn456").store_async()

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). Is is reccommended that name and download_as match to prevent confusion later on:

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.

    If no Name is specified this will be derived from the file name. This is the
    reccommended way to store a file in Synapse.

    Please note:
    The file, as it appears on disk, will be the file that is downloaded from
    Synapse. The name of the actual File is different from the name of the File
    Entity in Synapse. It is generally not reccommended to specify a different
    name for the Entity and the file as it will cause confusion and potential
    conflicts later on.

    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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The file object.

    Raises:
        ValueError: If the file does not have an ID and a path, or a path and a
            parent ID, or a data file handle ID and a parent ID.

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

        File with a parent and existing file handle (This allows multiple entities to reference the underlying file):

            file_instance = await File(data_file_handle_id="123", parent_id="syn456").store_async()

        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). Is is reccommended that `name` and
            `download_as` match to prevent confusion later on:

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

    """
    self.parent_id = parent.id if parent else self.parent_id
    if self._cannot_store():
        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)), or a "
            "(data_file_handle_id with a (`parent_id` or parent with an id)) to store."
        )
    self.name = self.name or (guess_file_name(self.path) if self.path else None)
    client = Synapse.get_client(synapse_client=synapse_client)

    if existing_file := await self._find_existing_file(synapse_client=client):
        merge_dataclass_entities(
            source=existing_file,
            destination=self,
            fields_to_ignore=self._determine_fields_to_ignore_in_merge(),
        )

    if self.path:
        self.path = os.path.expanduser(self.path)
        async with client._get_parallel_file_transfer_semaphore(
            asyncio_event_loop=asyncio.get_running_loop()
        ):
            await self._upload_file(synapse_client=client)
    elif self.data_file_handle_id:
        self.path = client.cache.get(file_handle_id=self.data_file_handle_id)

    if self.has_changed:
        synapse_file = Synapse_File(
            id=self.id,
            path=self.path,
            description=self.description,
            etag=self.etag,
            name=self.name,
            parent=parent.id if parent else self.parent_id,
            contentType=self.content_type,
            contentSize=self.content_size,
            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)

        entity = await store_entity(
            resource=self, entity=synapse_file, synapse_client=client
        )

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

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

    self._set_last_persistent_instance()

    client.logger.debug(f"Stored File {self.name}, id: {self.id}: {self.path}")
    # Clear the content_md5 so that it is recalculated if the file is updated
    self.content_md5 = None
    return self

copy_async async

copy_async(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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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.")
    from synapseutils.copy_functions import copy

    loop = asyncio.get_event_loop()

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

    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 async

delete_async(version_only: Optional[bool] = False, *, synapse_client: Optional[Synapse] = None) -> None

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
None	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        None

    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()
    await loop.run_in_executor(
        None,
        lambda: Synapse.get_client(synapse_client=synapse_client).delete(
            obj=self.id,
            version=self.version_number if version_only else None,
        ),
    )
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Deleted file {self.id}"
    )

from_id_async async classmethod

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

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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 async classmethod

from_path_async(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.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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 async

change_metadata_async(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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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.")
    from synapseutils.copy_functions import changeFileMetaData

    loop = asyncio.get_event_loop()

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

    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 async

get_permissions_async(*, synapse_client: Optional[Synapse] = None) -> Permissions

Get the permissions that the caller has on an Entity.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        A 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()

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

get_acl_async async

get_acl_async(principal_id: int = None, *, synapse_client: Optional[Synapse] = None) -> List[str]

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

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

set_permissions_async async

set_permissions_async(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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, list]]	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        An 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()

    return await loop.run_in_executor(
        None,
        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,
        ),
    )

---

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]
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. Activity cannot be removed during a store operation by setting it to None. You must use: synapseclient.models.Activity.delete_async or [synapseclient.models.Activity.disassociate_from_entity_async][]. TYPE: Optional[Activity]
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. To remove all annotations set this to an empty dict {} or None and store the entity. 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.
        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. Activity
            cannot be removed during a store operation by setting it to None. You must
            use: [synapseclient.models.Activity.delete_async][] or
            [synapseclient.models.Activity.disassociate_from_entity_async][].
        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. To remove all annotations set this
            to an empty dict `{}` or None and store the entity.

    """

    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. Activity cannot
    be removed during a store operation by setting it to None. You must use:
    [synapseclient.models.Activity.delete_async][] or
    [synapseclient.models.Activity.disassociate_from_entity_async][].
    """

    annotations: Optional[
        Dict[
            str,
            Union[
                List[str],
                List[bool],
                List[float],
                List[int],
                List[date],
                List[datetime],
            ],
        ]
    ] = field(default_factory=dict)
    """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 `{}`
    or None and store the entity."""

    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", {})
            )
        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The path to the CSV that was stored.
        """
        synapse_table = Synapse_Table(schema=self.id, values=csv_path)
        loop = asyncio.get_event_loop()
        entity = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).store(
                obj=synapse_table
            ),
        )
        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

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

    @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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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,
        )
        trace.get_current_span().set_attributes(
            {
                "synapse.name": self.name or "",
                "synapse.id": self.id or "",
            }
        )
        loop = asyncio.get_event_loop()
        entity = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).store(
                obj=synapse_schema
            ),
        )

        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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()
        entity = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).get(
                entity=self.id
            ),
        )
        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

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

    @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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The results of the query.
        """
        loop = asyncio.get_event_loop()

        # 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: Synapse.get_client(synapse_client=synapse_client).tableQuery(
                query=query,
                **result_format.to_dict(),
            ),
        )
        print(results)
        return results

Functions

get_async async

get_async(*, synapse_client: Optional[Synapse] = None) -> Table

Get the metadata about the table from synapse.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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()
    entity = await loop.run_in_executor(
        None,
        lambda: Synapse.get_client(synapse_client=synapse_client).get(
            entity=self.id
        ),
    )
    self.fill_from_dict(synapse_table=entity, set_annotations=True)
    return self

store_schema_async async

store_schema_async(*, synapse_client: Optional[Synapse] = None) -> Table

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

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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,
    )
    trace.get_current_span().set_attributes(
        {
            "synapse.name": self.name or "",
            "synapse.id": self.id or "",
        }
    )
    loop = asyncio.get_event_loop()
    entity = await loop.run_in_executor(
        None,
        lambda: Synapse.get_client(synapse_client=synapse_client).store(
            obj=synapse_schema
        ),
    )

    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 async

store_rows_from_csv_async(csv_path: str, *, synapse_client: Optional[Synapse] = None) -> str

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

delete_rows_async async

delete_rows_async(rows: List[Row], *, synapse_client: Optional[Synapse] = None) -> None

Delete rows from a table.

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

RETURNS	DESCRIPTION
None	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

query_async async classmethod

query_async(query: str, result_format: Union[CsvResultFormat, RowsetResultFormat] = CsvResultFormat(), *, synapse_client: Optional[Synapse] = None) -> Union[CsvFileTable, TableQueryResult]

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Union[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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The results of the query.
    """
    loop = asyncio.get_event_loop()

    # 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: Synapse.get_client(synapse_client=synapse_client).tableQuery(
            query=query,
            **result_format.to_dict(),
        ),
    )
    print(results)
    return results

delete_async async

delete_async(*, synapse_client: Optional[Synapse] = None) -> None

Delete the table from synapse.

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

RETURNS	DESCRIPTION
None	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

get_permissions_async async

get_permissions_async(*, synapse_client: Optional[Synapse] = None) -> Permissions

Get the permissions that the caller has on an Entity.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        A 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()

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

get_acl_async async

get_acl_async(principal_id: int = None, *, synapse_client: Optional[Synapse] = None) -> List[str]

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

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

set_permissions_async async

set_permissions_async(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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Union[str, list]]	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        An 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()

    return await loop.run_in_executor(
        None,
        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,
        ),
    )

---

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.
        """
        if not synapse_activity:
            synapse_activity = {}
        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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()
        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: Synapse.get_client(synapse_client=synapse_client).setProvenance(
                    entity=parent.id,
                    activity=synapse_activity,
                ),
            )
        else:
            saved_activity = await loop.run_in_executor(
                None,
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).updateActivity(
                    activity=synapse_activity,
                ),
            )
        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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()
            try:
                synapse_activity = await loop.run_in_executor(
                    None,
                    lambda: Synapse.get_client(
                        synapse_client=synapse_client
                    ).getProvenance(
                        entity=parent.id,
                        version=parent.version_number,
                    ),
                )
            except SynapseHTTPError as ex:
                if ex.response.status_code == 404:
                    return None
                else:
                    raise ex
            if synapse_activity:
                return cls().fill_from_dict(synapse_activity=synapse_activity)
            else:
                return None

    @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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        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()
            await loop.run_in_executor(
                None,
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).deleteProvenance(
                    entity=parent.id,
                ),
            )
            parent.activity = None

    @classmethod
    async def disassociate_from_entity_async(
        cls,
        parent: Union["Table", "File"],
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> None:
        """
        Disassociate the Activity from the parent entity. This is the first step in
        deleting the Activity. If you have other entities that are associated with this
        Activity you must disassociate them by calling this method on them as well.

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

        Raises:
            ValueError: If the parent does not have an ID.
        """
        # TODO: Input validation: SYNPY-1400
        with tracer.start_as_current_span(
            name=f"Activity_disassociate: Parent_ID: {parent.id}"
        ):
            await delete_entity_generated_by(
                entity_id=parent.id, synapse_client=synapse_client
            )
            parent.activity = None

Functions

from_parent_async async classmethod

from_parent_async(parent: Union[Table, File], *, synapse_client: Optional[Synapse] = None) -> Union[Activity, None]

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Union[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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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()
        try:
            synapse_activity = await loop.run_in_executor(
                None,
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).getProvenance(
                    entity=parent.id,
                    version=parent.version_number,
                ),
            )
        except SynapseHTTPError as ex:
            if ex.response.status_code == 404:
                return None
            else:
                raise ex
        if synapse_activity:
            return cls().fill_from_dict(synapse_activity=synapse_activity)
        else:
            return None

store_async async

store_async(parent: Optional[Union[Table, File]] = None, *, synapse_client: Optional[Synapse] = None) -> Activity

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The 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()
    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: Synapse.get_client(synapse_client=synapse_client).setProvenance(
                entity=parent.id,
                activity=synapse_activity,
            ),
        )
    else:
        saved_activity = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).updateActivity(
                activity=synapse_activity,
            ),
        )
    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 async classmethod

delete_async(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.

PARAMETER	DESCRIPTION
parent	The parent entity this activity is associated with. TYPE: Union[Table, File]
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RAISES	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    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()
        await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).deleteProvenance(
                entity=parent.id,
            ),
        )
        parent.activity = None

---

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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            Team: The Team object.
        """
        loop = asyncio.get_event_loop()
        trace.get_current_span().set_attributes(
            {
                "synapse.name": self.name or "",
                "synapse.id": self.id or "",
            }
        )
        team = await loop.run_in_executor(
            None,
            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,
            ),
        )
        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            None
        """
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).delete_team(
                id=self.id,
            ),
        )

    @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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        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()
            api_team = await loop.run_in_executor(
                None,
                lambda: Synapse.get_client(synapse_client=synapse_client).getTeam(
                    id=self.id,
                ),
            )
            return self.fill_from_dict(api_team)
        elif self.name:
            loop = asyncio.get_event_loop()
            api_team = await loop.run_in_executor(
                None,
                lambda: Synapse.get_client(synapse_client=synapse_client).getTeam(
                    id=self.name,
                ),
            )
            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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            List[TeamMember]: A List of TeamMember objects.
        """
        loop = asyncio.get_event_loop()
        team_members = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).getTeamMembers(
                team=self
            ),
        )
        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            dict: The invite response.
        """
        loop = asyncio.get_event_loop()
        invite = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).invite_to_team(
                team=self,
                user=user,
                message=message,
                force=force,
            ),
        )
        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            List[dict]: A list of invitations.
        """
        loop = asyncio.get_event_loop()
        invitations = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).get_team_open_invitations(
                team=self,
            ),
        )
        return list(invitations)

Functions

create_async async

create_async(*, synapse_client: Optional[Synapse] = None) -> Team

Creates a new team on Synapse.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        Team: The Team object.
    """
    loop = asyncio.get_event_loop()
    trace.get_current_span().set_attributes(
        {
            "synapse.name": self.name or "",
            "synapse.id": self.id or "",
        }
    )
    team = await loop.run_in_executor(
        None,
        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,
        ),
    )
    self.fill_from_dict(synapse_team=team)
    return self

delete_async async

delete_async(*, synapse_client: Optional[Synapse] = None) -> None

Deletes a team from Synapse.

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

RETURNS	DESCRIPTION
None	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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        None
    """
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(
        None,
        lambda: Synapse.get_client(synapse_client=synapse_client).delete_team(
            id=self.id,
        ),
    )

from_id_async async classmethod

from_id_async(id: int, *, synapse_client: Optional[Synapse] = None) -> Team

Gets Team object using its integer id.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        Team: The Team object.
    """

    return await cls(id=id).get_async(synapse_client=synapse_client)

from_name_async async classmethod

from_name_async(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. ***

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        Team: The Team object.
    """
    return await cls(name=name).get_async(synapse_client=synapse_client)

members_async async

members_async(*, synapse_client: Optional[Synapse] = None) -> List[TeamMember]

Gets the TeamMembers associated with a team given the ID field on the Team instance.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        List[TeamMember]: A List of TeamMember objects.
    """
    loop = asyncio.get_event_loop()
    team_members = await loop.run_in_executor(
        None,
        lambda: Synapse.get_client(synapse_client=synapse_client).getTeamMembers(
            team=self
        ),
    )
    team_member_list = [
        TeamMember().fill_from_dict(synapse_team_member=member)
        for member in team_members
    ]
    return team_member_list

invite_async async

invite_async(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.

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
dict	The 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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        dict: The invite response.
    """
    loop = asyncio.get_event_loop()
    invite = await loop.run_in_executor(
        None,
        lambda: Synapse.get_client(synapse_client=synapse_client).invite_to_team(
            team=self,
            user=user,
            message=message,
            force=force,
        ),
    )
    return invite

open_invitations_async async

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

Gets all open invitations for a team given the ID field on the Team instance.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        List[dict]: A list of invitations.
    """
    loop = asyncio.get_event_loop()
    invitations = await loop.run_in_executor(
        None,
        lambda: Synapse.get_client(
            synapse_client=synapse_client
        ).get_team_open_invitations(
            team=self,
        ),
    )
    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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The UserProfile object.

        """
        loop = asyncio.get_event_loop()

        if self.id:
            synapse_user_profile = await loop.run_in_executor(
                None,
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).get_user_profile_by_id(id=self.id),
            )
        elif self.username:
            synapse_user_profile = await loop.run_in_executor(
                None,
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).get_user_profile_by_username(username=self.username),
            )
        else:
            synapse_user_profile = await loop.run_in_executor(
                None,
                lambda: Synapse.get_client(
                    synapse_client=synapse_client
                ).get_user_profile_by_username(),
            )

        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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The 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 and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Returns:
            True if the user is certified, False otherwise.

        Raises:
            ValueError: If id nor username is specified.
        """
        loop = asyncio.get_event_loop()

        if self.id or self.username:
            is_certified = await loop.run_in_executor(
                None,
                lambda: Synapse.get_client(synapse_client=synapse_client).is_certified(
                    user=self.id or self.username
                ),
            )
        else:
            raise ValueError("Must specify either id or username")

        return is_certified

Functions

get_async async

get_async(*, 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.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The UserProfile object.

    """
    loop = asyncio.get_event_loop()

    if self.id:
        synapse_user_profile = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).get_user_profile_by_id(id=self.id),
        )
    elif self.username:
        synapse_user_profile = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).get_user_profile_by_username(username=self.username),
        )
    else:
        synapse_user_profile = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(
                synapse_client=synapse_client
            ).get_user_profile_by_username(),
        )

    self.fill_from_dict(synapse_user_profile=synapse_user_profile)
    return self

from_id_async async classmethod

from_id_async(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.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The UserProfile object.
    """

    return await cls(id=user_id).get_async(synapse_client=synapse_client)

from_username_async async classmethod

from_username_async(username: str, *, synapse_client: Optional[Synapse] = None) -> UserProfile

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 and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The UserProfile object.
    """
    return await cls(username=username).get_async(synapse_client=synapse_client)

is_certified_async async

is_certified_async(*, synapse_client: Optional[Synapse] = None) -> bool

Determine whether a user is certified.

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

RETURNS	DESCRIPTION
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 and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Returns:
        True if the user is certified, False otherwise.

    Raises:
        ValueError: If id nor username is specified.
    """
    loop = asyncio.get_event_loop()

    if self.id or self.username:
        is_certified = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).is_certified(
                user=self.id or self.username
            ),
        )
    else:
        raise ValueError("Must specify either id or username")

    return is_certified

---

synapseclient.models.Annotations dataclass

Bases: AnnotationsSynchronousProtocol

Annotations that can be applied to a number of Synapse resources to provide additional information.

ATTRIBUTE	DESCRIPTION
annotations	Additional metadata associated with the object. The key is the name of your desired annotations. The value is an object containing a list of string values (use empty list to represent no values for key) and the value type associated with all values in the list. TYPE: Union[Dict[str, Union[List[str], List[bool], List[float], List[int], List[date], List[datetime]]], None]
id	ID of the object to which this annotation belongs. Not required if being used as a member variable on another class. TYPE: Optional[str]
etag	Etag of the object to which this annotation belongs. This field must match the current etag on the object. Not required if being used as a member variable on another class. TYPE: Optional[str]

Source code in synapseclient/models/annotations.py

@dataclass()
@async_to_sync
class Annotations(AnnotationsSynchronousProtocol):
    """Annotations that can be applied to a number of Synapse resources to provide
    additional information.

    Attributes:
        annotations: Additional metadata associated with the object. The key is the name
            of your desired annotations. The value is an object containing a list of
            string values (use empty list to represent no values for key) and the value
            type associated with all values in the list.
        id: ID of the object to which this annotation belongs. Not required if being
            used as a member variable on another class.
        etag: Etag of the object to which this annotation belongs. This field must match
            the current etag on the object. Not required if being used as a member
            variable on another class.
    """

    annotations: Union[
        Dict[
            str,
            Union[
                List[str],
                List[bool],
                List[float],
                List[int],
                List[date],
                List[datetime],
            ],
        ],
        None,
    ] = field(default_factory=dict)
    """Additional metadata associated with the object. The key is the name of your
    desired annotations. The value is an object containing a list of string 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
    """ID of the object to which this annotation belongs. Not required if being used as
    a member variable on another class."""

    etag: Optional[str] = None
    """ Etag of the object to which this annotation belongs. To update an AnnotationV2,
    this field must match the current etag on the object. Not required if being used as
    a member variable on another class."""

    async def store_async(
        self,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> "Annotations":
        """Storing annotations to synapse.

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

        Returns:
            The stored annotations.

        Raises:
            ValueError: If the id or etag are not set.
        """
        if self.id is None or self.etag is None:
            raise ValueError("id and etag are required to store annotations.")

        result = await set_annotations_async(
            annotations=self,
            synapse_client=synapse_client,
        )
        self.annotations = Annotations.from_dict(result)
        self.etag = result["etag"]
        Synapse.get_client(synapse_client=synapse_client).logger.debug(
            f"Annotations stored for {self.id}"
        )
        return self

    @classmethod
    def from_dict(
        cls, synapse_annotations: dict
    ) -> Union[
        Dict[
            str,
            Union[
                List[str],
                List[bool],
                List[float],
                List[int],
                List[date],
                List[datetime],
            ],
        ],
        None,
    ]:
        """Convert the annotations from the format the synapse rest API works in -
        to the format used by this class.

        Arguments:
            synapse_annotations: The annotations from the synapse rest API.

        Returns:
            The annotations in python class format or None.
        """
        if synapse_annotations is None:
            return None
        annotations = {}
        dict_to_convert = (
            synapse_annotations["annotations"]
            if "annotations" in synapse_annotations
            else synapse_annotations
        )
        for key in dict_to_convert:
            if isinstance(dict_to_convert[key], dict):
                conversion_func = ANNO_TYPE_TO_FUNC[dict_to_convert[key]["type"]]
                annotations[key] = [
                    conversion_func(v) for v in dict_to_convert[key]["value"]
                ]
            else:
                annotations[key] = dict_to_convert[key]

        return annotations

Functions

store_async async

store_async(*, synapse_client: Optional[Synapse] = None) -> Annotations

Storing annotations to synapse.

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

RETURNS	DESCRIPTION
Annotations	The stored annotations.

RAISES	DESCRIPTION
ValueError	If the id or etag are not set.

Source code in synapseclient/models/annotations.py

async def store_async(
    self,
    *,
    synapse_client: Optional[Synapse] = None,
) -> "Annotations":
    """Storing annotations to synapse.

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

    Returns:
        The stored annotations.

    Raises:
        ValueError: If the id or etag are not set.
    """
    if self.id is None or self.etag is None:
        raise ValueError("id and etag are required to store annotations.")

    result = await set_annotations_async(
        annotations=self,
        synapse_client=synapse_client,
    )
    self.annotations = Annotations.from_dict(result)
    self.etag = result["etag"]
    Synapse.get_client(synapse_client=synapse_client).logger.debug(
        f"Annotations stored for {self.id}"
    )
    return self