Curator

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.

API reference

synapseclient.models.CurationTask dataclass

Bases: CurationTaskSynchronousProtocol

The CurationTask provides instructions for a Data Contributor on how data or metadata of a specific type should be both added to a project and curated.

Represents a Synapse CurationTask.

ATTRIBUTE	DESCRIPTION
task_id	The unique identifier issued to this task when it was created TYPE: Optional[int]
data_type	Will match the data type that a contributor plans to contribute TYPE: Optional[str]
project_id	The synId of the project TYPE: Optional[str]
instructions	Instructions to the data contributor TYPE: Optional[str]
task_properties	The properties of a CurationTask. This can be either FileBasedMetadataTaskProperties or RecordBasedMetadataTaskProperties. TYPE: Optional[Union[FileBasedMetadataTaskProperties, RecordBasedMetadataTaskProperties]]
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	(Read Only) The date this task was created TYPE: Optional[str]
modified_on	(Read Only) The date this task was last modified TYPE: Optional[str]
created_by	(Read Only) The ID of the user that created this task TYPE: Optional[str]
modified_by	(Read Only) The ID of the user that last modified this task TYPE: Optional[str]

Complete curation task workflow

 

from synapseclient import Synapse
from synapseclient.models import CurationTask, FileBasedMetadataTaskProperties

syn = Synapse()
syn.login()

# Create a new file-based curation task
file_properties = FileBasedMetadataTaskProperties(
    upload_folder_id="syn1234567",
    file_view_id="syn2345678"
)

task = CurationTask(
    project_id="syn9876543",
    data_type="genomics_data",
    instructions="Upload your genomics files and complete metadata",
    task_properties=file_properties
)
task = task.store()
print(f"Created task: {task.task_id}")

# Later, retrieve and update the task
existing_task = CurationTask(task_id=task.task_id).get()
existing_task.instructions = "Updated instructions with new requirements"
existing_task.store()

# List all tasks in the project
for project_task in CurationTask.list(project_id="syn9876543"):
    print(f"Task: {project_task.data_type} - {project_task.task_id}")

Source code in synapseclient/models/curation.py

@dataclass
@async_to_sync
class CurationTask(CurationTaskSynchronousProtocol):
    """
    The CurationTask provides instructions for a Data Contributor on how data or metadata
    of a specific type should be both added to a project and curated.

    Represents a [Synapse CurationTask](https://rest-docs.synapse.org/org/sagebionetworks/repo/model/curation/CurationTask.html).

    Attributes:
        task_id: The unique identifier issued to this task when it was created
        data_type: Will match the data type that a contributor plans to contribute
        project_id: The synId of the project
        instructions: Instructions to the data contributor
        task_properties: The properties of a CurationTask. This can be either
            FileBasedMetadataTaskProperties or RecordBasedMetadataTaskProperties.
        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: (Read Only) The date this task was created
        modified_on: (Read Only) The date this task was last modified
        created_by: (Read Only) The ID of the user that created this task
        modified_by: (Read Only) The ID of the user that last modified this task

    Example: Complete curation task workflow
         

        ```python
        from synapseclient import Synapse
        from synapseclient.models import CurationTask, FileBasedMetadataTaskProperties

        syn = Synapse()
        syn.login()

        # Create a new file-based curation task
        file_properties = FileBasedMetadataTaskProperties(
            upload_folder_id="syn1234567",
            file_view_id="syn2345678"
        )

        task = CurationTask(
            project_id="syn9876543",
            data_type="genomics_data",
            instructions="Upload your genomics files and complete metadata",
            task_properties=file_properties
        )
        task = task.store()
        print(f"Created task: {task.task_id}")

        # Later, retrieve and update the task
        existing_task = CurationTask(task_id=task.task_id).get()
        existing_task.instructions = "Updated instructions with new requirements"
        existing_task.store()

        # List all tasks in the project
        for project_task in CurationTask.list(project_id="syn9876543"):
            print(f"Task: {project_task.data_type} - {project_task.task_id}")
        ```
    """

    task_id: Optional[int] = None
    """The unique identifier issued to this task when it was created"""

    data_type: Optional[str] = None
    """Will match the data type that a contributor plans to contribute. The dataType must be unique within a project"""

    project_id: Optional[str] = None
    """The synId of the project"""

    instructions: Optional[str] = None
    """Instructions to the data contributor"""

    task_properties: Optional[
        Union[FileBasedMetadataTaskProperties, RecordBasedMetadataTaskProperties]
    ] = None
    """The properties of a CurationTask"""

    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
    """(Read Only) The date this task was created"""

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

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

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

    assignee_principal_id: Optional[str] = None
    """The principal ID of the user or team assigned to this task. Null if unassigned. For metadata
    tasks, determines the owner of the grid session. Team members can all join grid sessions
    owned by their team, while user-owned grid sessions are restricted to that user only."""

    _last_persistent_instance: Optional["CurationTask"] = 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.task_properties = (
            deepcopy(self.task_properties) if self.task_properties else None
        )

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

        Arguments:
            synapse_response: The response from the REST API.

        Returns:
            The CurationTask object.
        """
        self.task_id = (
            int(synapse_response.get("taskId", None))
            if synapse_response.get("taskId", None)
            else None
        )
        self.data_type = synapse_response.get("dataType", None)
        self.project_id = synapse_response.get("projectId", None)
        self.instructions = synapse_response.get("instructions", None)
        self.etag = synapse_response.get("etag", None)
        self.created_on = synapse_response.get("createdOn", None)
        self.modified_on = synapse_response.get("modifiedOn", None)
        self.created_by = synapse_response.get("createdBy", None)
        self.modified_by = synapse_response.get("modifiedBy", None)
        self.assignee_principal_id = synapse_response.get("assigneePrincipalId", None)

        task_properties_dict = synapse_response.get("taskProperties", None)
        if task_properties_dict is None:
            raise ValueError(
                "taskProperties was not found in the Synapse response for this CurationTask. "
                "This means it is likely an older CurationTask from before taskProperties was added. "
                "It is recommended that this task be deleted: task.delete(delete_source=False) "
                "and then recreate the task with the correct taskProperties."
            )
        self.task_properties = _create_task_properties_from_dict(task_properties_dict)

        return self

    def to_synapse_request(self) -> Dict[str, Any]:
        """
        Converts this dataclass to a dictionary suitable for a Synapse REST API request.

        Returns:
            A dictionary representation of this object for API requests.
        """
        request_dict = {}
        request_dict["taskId"] = self.task_id
        request_dict["dataType"] = self.data_type
        request_dict["projectId"] = self.project_id
        request_dict["instructions"] = self.instructions
        request_dict["etag"] = self.etag
        request_dict["createdOn"] = self.created_on
        request_dict["modifiedOn"] = self.modified_on
        request_dict["createdBy"] = self.created_by
        request_dict["modifiedBy"] = self.modified_by
        request_dict["assigneePrincipalId"] = self.assignee_principal_id

        if self.task_properties is not None:
            request_dict["taskProperties"] = self.task_properties.to_synapse_request()

        delete_none_keys(request_dict)
        return request_dict

    async def get_async(
        self, *, synapse_client: Optional[Synapse] = None
    ) -> "CurationTask":
        """
        Gets a CurationTask from Synapse by 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:
            CurationTask: The CurationTask object.

        Raises:
            ValueError: If the CurationTask object does not have a task_id.
            ValueError: If the Synapse response does not contain taskProperties.

        Example: Get a curation task asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask

            syn = Synapse()
            syn.login()

            async def main():
                task = await CurationTask(task_id=123).get_async()
                print(f"Data type: {task.data_type}")
                print(f"Instructions: {task.instructions}")

            asyncio.run(main())
            ```
        """
        if not self.task_id:
            raise ValueError("task_id is required to get a CurationTask")

        trace.get_current_span().set_attributes(
            {
                "synapse.task_id": str(self.task_id),
            }
        )

        task_result = await get_curation_task(
            task_id=self.task_id, synapse_client=synapse_client
        )
        self.fill_from_dict(synapse_response=task_result)
        self._set_last_persistent_instance()
        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: (
            f"CurationTask_GetStatus: ID: {self.task_id}"
        )
    )
    async def get_status_async(
        self, *, synapse_client: Synapse | None = None
    ) -> "CurationTaskStatus":
        """
        Gets the status of this CurationTask 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 CurationTaskStatus object.

        Raises:
            ValueError: If the CurationTask object does not have a task_id.

        Example: Get the status of a curation task asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask

            syn = Synapse()
            syn.login()

            async def main():
                status = await CurationTask(task_id=123).get_status_async()
                print(status.state)

            asyncio.run(main())
            ```
        """
        if not self.task_id:
            raise ValueError("task_id is required to get a CurationTask status")

        status_result = await get_curation_task_status(
            task_id=self.task_id, synapse_client=synapse_client
        )
        return CurationTaskStatus().fill_from_dict(status_result)

    async def delete_async(
        self,
        delete_source: bool = False,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> None:
        """
        Deletes a CurationTask from Synapse.

        Arguments:
            delete_source: If True, the associated source data (EntityView or RecordSet) will also be deleted
                if the task is a FileBasedMetadataTask or RecordBasedMetadataTask respectively. Defaults to False.
            synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Raises:
            ValueError: If the CurationTask object does not have a task_id.
            ValueError: If delete_source is True but the task properties are not properly set
              to identify the source to delete.

        Example: Delete a curation task asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask

            syn = Synapse()
            syn.login()

            async def main():
                task = CurationTask(task_id=123)
                await task.delete_async()
                print("Task deleted successfully")

            asyncio.run(main())
            ```

        Example: Delete a curation task and its associated data source asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask

            syn = Synapse()
            syn.login()

            async def main():
                task = CurationTask(task_id=123)
                await task.delete_async(delete_source=True)
                print("Task and record set deleted successfully")

            asyncio.run(main())
            ```
        """
        if not self.task_id:
            raise ValueError("task_id is required to delete a CurationTask")

        trace.get_current_span().set_attributes(
            {
                "synapse.task_id": str(self.task_id),
            }
        )

        if delete_source:
            if not self.task_properties:
                await self.get_async(synapse_client=synapse_client)

            if isinstance(self.task_properties, FileBasedMetadataTaskProperties):
                if not self.task_properties.file_view_id:
                    raise ValueError(
                        "Cannot delete Fileview: "
                        "'file_view_id' attribute is missing."
                    )
                from synapseclient.models import EntityView

                await EntityView(id=self.task_properties.file_view_id).delete_async(
                    synapse_client=synapse_client
                )

            elif isinstance(self.task_properties, RecordBasedMetadataTaskProperties):
                if not self.task_properties.record_set_id:
                    raise ValueError(
                        "Cannot delete RecordSet: "
                        "'record_set_id' attribute is missing."
                    )
                from synapseclient.models import RecordSet

                await RecordSet(id=self.task_properties.record_set_id).delete_async(
                    synapse_client=synapse_client
                )

            else:
                raise ValueError(
                    "'task_property' attribute is None. "
                    "Deletion only supports FileBasedMetadataTaskProperties or "
                    "RecordBasedMetadataTaskProperties."
                )

        await delete_curation_task(task_id=self.task_id, synapse_client=synapse_client)

    async def store_async(
        self, *, synapse_client: Optional[Synapse] = None
    ) -> "CurationTask":
        """
        Creates a new CurationTask or updates an existing one on Synapse.

        This method implements non-destructive updates. If a CurationTask with the same
        project_id and data_type exists and this instance hasn't been retrieved from
        Synapse before, it will merge the existing task data with the current instance
        before updating.

        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:
            CurationTask: The CurationTask object.

        Example: Create a new curation task asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask, FileBasedMetadataTaskProperties

            syn = Synapse()
            syn.login()

            async def main():
                # Create file-based task properties
                file_properties = FileBasedMetadataTaskProperties(
                    upload_folder_id="syn1234567",
                    file_view_id="syn2345678"
                )

                # Create and store the curation task
                task = CurationTask(
                    project_id="syn9876543",
                    data_type="genomics_data",
                    instructions="Upload your genomics files to the specified folder",
                    task_properties=file_properties
                )
                task = await task.store_async()
                print(f"Created task with ID: {task.task_id}")

            asyncio.run(main())
            ```
        """
        if not self.project_id:
            raise ValueError("project_id is required")
        if not self.data_type:
            raise ValueError("data_type is required")

        trace.get_current_span().set_attributes(
            {
                "synapse.data_type": self.data_type or "",
                "synapse.project_id": self.project_id or "",
                "synapse.task_id": str(self.task_id) if self.task_id else "",
            }
        )

        if (
            not self._last_persistent_instance
            and not self.task_id
            and (
                existing_task_id := await _get_existing_curation_task_id(
                    project_id=self.project_id,
                    data_type=self.data_type,
                    synapse_client=synapse_client,
                )
            )
            and (
                existing_task := await CurationTask(task_id=existing_task_id).get_async(
                    synapse_client=synapse_client
                )
            )
        ):
            merge_dataclass_entities(source=existing_task, destination=self)

        if self.task_id:
            task_result = await update_curation_task(
                task_id=self.task_id,
                curation_task=self.to_synapse_request(),
                synapse_client=synapse_client,
            )
            self.fill_from_dict(synapse_response=task_result)
            self._set_last_persistent_instance()
            return self
        else:
            if not self.project_id:
                raise ValueError("project_id is required to create a CurationTask")
            if not self.data_type:
                raise ValueError("data_type is required to create a CurationTask")
            if not self.instructions:
                raise ValueError("instructions is required to create a CurationTask")
            if not self.task_properties:
                raise ValueError("task_properties is required to create a CurationTask")

            task_result = await create_curation_task(
                curation_task=self.to_synapse_request(),
                synapse_client=synapse_client,
            )
            self.fill_from_dict(synapse_response=task_result)
            self._set_last_persistent_instance()
            return self

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: (
            f"CurationTask_UpdateStatus: ID: {self.task_id}"
        )
    )
    async def update_status_async(
        self,
        curation_task_status: "CurationTaskStatus",
        *,
        synapse_client: Synapse | None = None,
    ) -> "CurationTaskStatus":
        """
        Updates the status of this CurationTask on Synapse.

        Arguments:
            curation_task_status: The complete CurationTaskStatus object to update.
            synapse_client: If not passed in and caching was not disabled by
                Synapse.allow_client_caching(False) this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The updated CurationTaskStatus object.

        Raises:
            ValueError: If the CurationTask object does not have a task_id.

        Example: Update the status of a curation task asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import (
                CurationTask,
                TaskState,
            )

            syn = Synapse()
            syn.login()

            async def main():
                task = CurationTask(task_id=123)
                current = await task.get_status_async()
                current.state = TaskState.COMPLETED
                updated = await task.update_status_async(curation_task_status=current)
                print(updated.state)

            asyncio.run(main())
            ```
        """
        if not self.task_id:
            raise ValueError("task_id is required to update a CurationTask status")

        status_result = await update_curation_task_status(
            task_id=self.task_id,
            curation_task_status=curation_task_status.to_synapse_request(),
            synapse_client=synapse_client,
        )
        return CurationTaskStatus().fill_from_dict(status_result)

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: (
            f"CurationTask_SetActiveGridSession: ID: {self.task_id}"
        )
    )
    async def set_active_grid_session_async(
        self,
        active_session_id: str,
        *,
        synapse_client: Synapse | None = None,
    ) -> "CurationTaskStatus":
        """
        Set the active grid session on this CurationTask's status by replacing
        execution_details with a GridExecutionDetails carrying the given session id.

        Arguments:
            active_session_id: The unique identifier of the active grid session to link.
            synapse_client: If not passed in and caching was not disabled by
                Synapse.allow_client_caching(False) this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The updated CurationTaskStatus object.

        Raises:
            ValueError: If the CurationTask object does not have a task_id.

        Example: Link a grid session to a curation task asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask, Grid

            syn = Synapse()
            syn.login()

            async def main():
                grid = await Grid(record_set_id="syn1234567").create_async()
                await CurationTask(task_id=123).set_active_grid_session_async(
                    active_session_id=grid.session_id
                )

            asyncio.run(main())
            ```
        """
        status = await self.get_status_async(synapse_client=synapse_client)
        status.execution_details = GridExecutionDetails(
            active_session_id=active_session_id
        )
        return await self.update_status_async(
            curation_task_status=status, synapse_client=synapse_client
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: (
            f"CurationTask_SetTaskState: ID: {self.task_id}"
        )
    )
    async def set_task_state_async(
        self,
        state: "TaskState | str",
        *,
        synapse_client: Synapse | None = None,
    ) -> "CurationTaskStatus":
        """
        Set the state on this CurationTask's status.

        Does not modify execution_details. Fetches the current CurationTaskStatus
        first so the update carries a fresh etag.

        Arguments:
            state: The state to set on this task's status. Accepts a
                TaskState or a string exactly matching one of its members
                (e.g. NOT_STARTED, IN_PROGRESS, COMPLETED, CANCELED).
            synapse_client: If not passed in and caching was not disabled by
                Synapse.allow_client_caching(False) this will use the last created
                instance from the Synapse class constructor.

        Returns:
            The updated CurationTaskStatus object.

        Raises:
            ValueError: If the CurationTask object does not have a task_id, or
                if state is a string that does not match a TaskState member.

        Example: Mark a curation task as completed asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask, TaskState

            syn = Synapse()
            syn.login()

            async def main():
                await CurationTask(task_id=123).set_task_state_async(
                    state=TaskState.COMPLETED
                )

            asyncio.run(main())
            ```

        Example: Mark a curation task as completed using a string asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask

            syn = Synapse()
            syn.login()

            async def main():
                await CurationTask(task_id=123).set_task_state_async(
                    state="COMPLETED"
                )

            asyncio.run(main())
            ```
        """
        try:
            coerced_state = TaskState(state)
        except ValueError as exc:
            raise ValueError(
                f"{state!r} is not a valid TaskState. "
                f"Expected one of: {[s.value for s in TaskState]}."
            ) from exc

        status = await self.get_status_async(synapse_client=synapse_client)
        status.state = coerced_state
        return await self.update_status_async(
            curation_task_status=status, synapse_client=synapse_client
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: (
            f"CurationTask_CreateGridSession: ID: {self.task_id}"
        )
    )
    async def create_grid_session_async(
        self,
        *,
        owner_principal_id: int | None = None,
        timeout: int = 120,
        synapse_client: Synapse | None = None,
    ) -> "Grid":
        """
        Create a new Grid session for this CurationTask and set it as the active session.

        Picks the Grid seed from this task's task_properties:

        - RecordBasedMetadataTaskProperties uses record_set_id
        - FileBasedMetadataTaskProperties uses an initial_query that selects from
          the file_view_id

        Always creates a new Grid session. To attach an existing session to a task,
        use set_active_grid_session_async instead.

        After the Grid is created, updates the CurationTaskStatus to point its
        active_session_id at the new session. If that update fails for any reason,
        the newly created Grid is deleted on a best-effort basis and the original
        exception is re-raised.

        Arguments:
            owner_principal_id: The principal ID (user or team) that will own the
                created grid session. When not provided, the principal ID of the
                caller is used.
            timeout: Seconds to wait for the grid creation job. Defaults to 120.
            synapse_client: If not passed 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 newly created Grid.

        Raises:
            ValueError: If task_id is unset or task_properties is of an unsupported type.
            SynapseHTTPError: If the RecordSet or EntityView does not exist, or if the
                status update fails. The orphan Grid is deleted on a best-effort basis
                before the error is re-raised.

        Example: Create a grid session for a curation task asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask

            syn = Synapse()
            syn.login()

            async def main():
                grid = await CurationTask(task_id=123).create_grid_session_async()
                print(grid.session_id)

            asyncio.run(main())
            ```
        """
        if not self.task_id:
            raise ValueError(
                "task_id is required to create a CurationTask grid session"
            )

        if not self.task_properties:
            await self.get_async(synapse_client=synapse_client)

        if isinstance(self.task_properties, RecordBasedMetadataTaskProperties):
            if not self.task_properties.record_set_id:
                raise ValueError(
                    "Cannot create grid session: "
                    "task_properties.record_set_id is missing."
                )
            from synapseclient.models import RecordSet

            # raises SynapseHTTPError if RecordSet does not exist
            await RecordSet(id=self.task_properties.record_set_id).get_async(
                synapse_client=synapse_client
            )
            grid = Grid(
                record_set_id=self.task_properties.record_set_id,
                owner_principal_id=owner_principal_id,
            )
        elif isinstance(self.task_properties, FileBasedMetadataTaskProperties):
            if not self.task_properties.file_view_id:
                raise ValueError(
                    "Cannot create grid session: "
                    "task_properties.file_view_id is missing."
                )
            from synapseclient.models import EntityView

            # raises SynapseHTTPError if EntityView does not exist
            await EntityView(id=self.task_properties.file_view_id).get_async(
                synapse_client=synapse_client
            )
            grid = Grid(
                initial_query=Query(
                    sql=f"SELECT * FROM {self.task_properties.file_view_id}"
                ),
                owner_principal_id=owner_principal_id,
            )
        else:
            raise ValueError(
                "task_properties must be a FileBasedMetadataTaskProperties or "
                "RecordBasedMetadataTaskProperties to create a grid session"
            )

        grid = await grid.create_async(
            timeout=timeout,
            synapse_client=synapse_client,
        )

        # Only one grid session can be set as the active one on a a given CurationTask
        # at a any time, though multiple sessions can exist.
        # If two users run this concurrently, one will lose the race and
        # receive a 412 (precondition failed). In that case — or if recording the
        # active session fails for any other reason — delete the session we just
        # created so it doesn't become an orphan. If the delete also fails, log a
        # warning so the caller knows manual cleanup is needed, then re-raise the
        # original exception in all cases.
        try:
            await self.set_active_grid_session_async(
                active_session_id=grid.session_id, synapse_client=synapse_client
            )
        except Exception:
            try:
                await grid.delete_async(synapse_client=synapse_client)
            except Exception:
                Synapse.get_client(synapse_client=synapse_client).logger.warning(
                    "Failed to delete orphan grid session %s after status "
                    "update failure; manual cleanup may be required.",
                    grid.session_id,
                )
            raise

        return grid

    @skip_async_to_sync
    @classmethod
    async def list_async(
        cls,
        project_id: str,
        *,
        assigned_to_me: Optional[bool] = None,
        assignee_ids: Optional[list[str]] = None,
        state_filter: Optional[list[Union["TaskState", str]]] = None,
        synapse_client: Optional[Synapse] = None,
    ) -> AsyncGenerator["CurationTask", None]:
        """
        Generator that yields CurationTasks for a project as they become available.

        Arguments:
            project_id: The synId of the project.
            assigned_to_me: When True, only return tasks assigned to the current user.
                Cannot be combined with assignee_ids.
                False does not mean "tasks not assigned to me".
                Defaults to None.
            assignee_ids: Optional list of principal IDs (users or teams) to filter
                tasks by assignee. Cannot be combined with assigned_to_me=True.
                Passing an empty list raises a ValueError; pass None to return tasks
                for any assignee. Defaults to None.
            state_filter: Optional list of TaskState values or exact-case strings to
                filter tasks by their current state (e.g., "IN_PROGRESS"). Defaults to
                None (all states returned). Passing an empty list raises a ValueError;
                pass None to return tasks in any state.
            synapse_client: If not passed in and caching was not disabled by
                Synapse.allow_client_caching(False) this will use the last created
                instance from the Synapse class constructor.

        Yields:
            CurationTask objects as they are retrieved from the API.

        Raises:
            ValueError: If state_filter is an empty list.
            ValueError: If assignee_ids is an empty list.
            ValueError: If assigned_to_me is True and assignee_ids is also provided.
            ValueError: If any value in state_filter is not a TaskState member or
                an exact-case string matching a TaskState value (e.g., "IN_PROGRESS").
            ValueError: If the Synapse response for any task does not contain
                taskProperties.

        Example: List all curation tasks in a project asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask

            syn = Synapse()
            syn.login()

            async def main():
                # List all curation tasks in the project
                async for task in CurationTask.list_async(project_id="syn9876543"):
                    print(f"Task ID: {task.task_id}")
                    print(f"Data Type: {task.data_type}")
                    print(f"Instructions: {task.instructions}")
                    print("---")

            asyncio.run(main())
            ```

        Example: List only curation tasks assigned to the current user asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask

            syn = Synapse()
            syn.login()

            async def main():
                async for task in CurationTask.list_async(
                    project_id="syn9876543", assigned_to_me=True
                ):
                    print(f"Task ID: {task.task_id}")
                    print(f"Data Type: {task.data_type}")
                    print("---")

            asyncio.run(main())
            ```

        Example: List only in-progress curation tasks asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask, TaskState

            syn = Synapse()
            syn.login()

            async def main():
                async for task in CurationTask.list_async(
                    project_id="syn9876543",
                    state_filter=[TaskState.IN_PROGRESS],
                ):
                    print(f"Task ID: {task.task_id}")
                    print(f"Data Type: {task.data_type}")
                    print("---")

            asyncio.run(main())
            ```

        Example: List only in-progress curation tasks using a string state filter asynchronously
             

            state_filter also accepts plain strings matching TaskState names exactly.

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import CurationTask

            syn = Synapse()
            syn.login()

            async def main():
                async for task in CurationTask.list_async(
                    project_id="syn9876543",
                    state_filter=["IN_PROGRESS"],
                ):
                    print(f"Task ID: {task.task_id}")
                    print(f"Data Type: {task.data_type}")
                    print("---")

            asyncio.run(main())
            ```
        """
        if state_filter == []:
            raise ValueError(
                "state_filter must not be empty. Pass None to return tasks in any state."
            )
        if assignee_ids == []:
            raise ValueError(
                "assignee_ids must not be empty. Pass None to return tasks for any assignee."
            )
        if assigned_to_me is True and assignee_ids is not None:
            raise ValueError(
                f"assigned_to_me and assignee_ids are mutually exclusive "
                f"and cannot be used together. Got assignee_ids={assignee_ids!r}."
            )

        if state_filter is not None:
            state_filter = coerce_enum_list(TaskState, state_filter)

        trace.get_current_span().set_attributes(
            {
                "synapse.project_id": project_id,
            }
        )

        async for task_dict in list_curation_tasks(
            project_id=project_id,
            assigned_to_me=assigned_to_me,
            assignee_ids=assignee_ids,
            state_filter=state_filter,
            synapse_client=synapse_client,
        ):
            task = cls().fill_from_dict(synapse_response=task_dict)
            yield task

Methods:

get_async async

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

Gets a CurationTask from Synapse by 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
CurationTask	The CurationTask object. TYPE: CurationTask

RAISES	DESCRIPTION
ValueError	If the CurationTask object does not have a task_id.
ValueError	If the Synapse response does not contain taskProperties.

Get a curation task asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask

syn = Synapse()
syn.login()

async def main():
    task = await CurationTask(task_id=123).get_async()
    print(f"Data type: {task.data_type}")
    print(f"Instructions: {task.instructions}")

asyncio.run(main())

Source code in synapseclient/models/curation.py

async def get_async(
    self, *, synapse_client: Optional[Synapse] = None
) -> "CurationTask":
    """
    Gets a CurationTask from Synapse by 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:
        CurationTask: The CurationTask object.

    Raises:
        ValueError: If the CurationTask object does not have a task_id.
        ValueError: If the Synapse response does not contain taskProperties.

    Example: Get a curation task asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask

        syn = Synapse()
        syn.login()

        async def main():
            task = await CurationTask(task_id=123).get_async()
            print(f"Data type: {task.data_type}")
            print(f"Instructions: {task.instructions}")

        asyncio.run(main())
        ```
    """
    if not self.task_id:
        raise ValueError("task_id is required to get a CurationTask")

    trace.get_current_span().set_attributes(
        {
            "synapse.task_id": str(self.task_id),
        }
    )

    task_result = await get_curation_task(
        task_id=self.task_id, synapse_client=synapse_client
    )
    self.fill_from_dict(synapse_response=task_result)
    self._set_last_persistent_instance()
    return self

get_status_async async

get_status_async(*, synapse_client: Synapse | None = None) -> CurationTaskStatus

Gets the status of this CurationTask 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: Synapse | None DEFAULT: None

RETURNS	DESCRIPTION
CurationTaskStatus	The CurationTaskStatus object.

RAISES	DESCRIPTION
ValueError	If the CurationTask object does not have a task_id.

Get the status of a curation task asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask

syn = Synapse()
syn.login()

async def main():
    status = await CurationTask(task_id=123).get_status_async()
    print(status.state)

asyncio.run(main())

Source code in synapseclient/models/curation.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: (
        f"CurationTask_GetStatus: ID: {self.task_id}"
    )
)
async def get_status_async(
    self, *, synapse_client: Synapse | None = None
) -> "CurationTaskStatus":
    """
    Gets the status of this CurationTask 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 CurationTaskStatus object.

    Raises:
        ValueError: If the CurationTask object does not have a task_id.

    Example: Get the status of a curation task asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask

        syn = Synapse()
        syn.login()

        async def main():
            status = await CurationTask(task_id=123).get_status_async()
            print(status.state)

        asyncio.run(main())
        ```
    """
    if not self.task_id:
        raise ValueError("task_id is required to get a CurationTask status")

    status_result = await get_curation_task_status(
        task_id=self.task_id, synapse_client=synapse_client
    )
    return CurationTaskStatus().fill_from_dict(status_result)

update_status_async async

update_status_async(curation_task_status: CurationTaskStatus, *, synapse_client: Synapse | None = None) -> CurationTaskStatus

Updates the status of this CurationTask on Synapse.

PARAMETER	DESCRIPTION
curation_task_status	The complete CurationTaskStatus object to update. TYPE: CurationTaskStatus
synapse_client	If not passed 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: Synapse | None DEFAULT: None

RETURNS	DESCRIPTION
CurationTaskStatus	The updated CurationTaskStatus object.

RAISES	DESCRIPTION
ValueError	If the CurationTask object does not have a task_id.

Update the status of a curation task asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import (
    CurationTask,
    TaskState,
)

syn = Synapse()
syn.login()

async def main():
    task = CurationTask(task_id=123)
    current = await task.get_status_async()
    current.state = TaskState.COMPLETED
    updated = await task.update_status_async(curation_task_status=current)
    print(updated.state)

asyncio.run(main())

Source code in synapseclient/models/curation.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: (
        f"CurationTask_UpdateStatus: ID: {self.task_id}"
    )
)
async def update_status_async(
    self,
    curation_task_status: "CurationTaskStatus",
    *,
    synapse_client: Synapse | None = None,
) -> "CurationTaskStatus":
    """
    Updates the status of this CurationTask on Synapse.

    Arguments:
        curation_task_status: The complete CurationTaskStatus object to update.
        synapse_client: If not passed in and caching was not disabled by
            Synapse.allow_client_caching(False) this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The updated CurationTaskStatus object.

    Raises:
        ValueError: If the CurationTask object does not have a task_id.

    Example: Update the status of a curation task asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import (
            CurationTask,
            TaskState,
        )

        syn = Synapse()
        syn.login()

        async def main():
            task = CurationTask(task_id=123)
            current = await task.get_status_async()
            current.state = TaskState.COMPLETED
            updated = await task.update_status_async(curation_task_status=current)
            print(updated.state)

        asyncio.run(main())
        ```
    """
    if not self.task_id:
        raise ValueError("task_id is required to update a CurationTask status")

    status_result = await update_curation_task_status(
        task_id=self.task_id,
        curation_task_status=curation_task_status.to_synapse_request(),
        synapse_client=synapse_client,
    )
    return CurationTaskStatus().fill_from_dict(status_result)

set_active_grid_session_async async

set_active_grid_session_async(active_session_id: str, *, synapse_client: Synapse | None = None) -> CurationTaskStatus

Set the active grid session on this CurationTask's status by replacing execution_details with a GridExecutionDetails carrying the given session id.

PARAMETER	DESCRIPTION
active_session_id	The unique identifier of the active grid session to link. 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: Synapse | None DEFAULT: None

RETURNS	DESCRIPTION
CurationTaskStatus	The updated CurationTaskStatus object.

RAISES	DESCRIPTION
ValueError	If the CurationTask object does not have a task_id.

Link a grid session to a curation task asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask, Grid

syn = Synapse()
syn.login()

async def main():
    grid = await Grid(record_set_id="syn1234567").create_async()
    await CurationTask(task_id=123).set_active_grid_session_async(
        active_session_id=grid.session_id
    )

asyncio.run(main())

Source code in synapseclient/models/curation.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: (
        f"CurationTask_SetActiveGridSession: ID: {self.task_id}"
    )
)
async def set_active_grid_session_async(
    self,
    active_session_id: str,
    *,
    synapse_client: Synapse | None = None,
) -> "CurationTaskStatus":
    """
    Set the active grid session on this CurationTask's status by replacing
    execution_details with a GridExecutionDetails carrying the given session id.

    Arguments:
        active_session_id: The unique identifier of the active grid session to link.
        synapse_client: If not passed in and caching was not disabled by
            Synapse.allow_client_caching(False) this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The updated CurationTaskStatus object.

    Raises:
        ValueError: If the CurationTask object does not have a task_id.

    Example: Link a grid session to a curation task asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask, Grid

        syn = Synapse()
        syn.login()

        async def main():
            grid = await Grid(record_set_id="syn1234567").create_async()
            await CurationTask(task_id=123).set_active_grid_session_async(
                active_session_id=grid.session_id
            )

        asyncio.run(main())
        ```
    """
    status = await self.get_status_async(synapse_client=synapse_client)
    status.execution_details = GridExecutionDetails(
        active_session_id=active_session_id
    )
    return await self.update_status_async(
        curation_task_status=status, synapse_client=synapse_client
    )

delete_async async

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

Deletes a CurationTask from Synapse.

PARAMETER	DESCRIPTION
delete_source	If True, the associated source data (EntityView or RecordSet) will also be deleted if the task is a FileBasedMetadataTask or RecordBasedMetadataTask respectively. Defaults to False. TYPE: bool DEFAULT: False
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

RAISES	DESCRIPTION
ValueError	If the CurationTask object does not have a task_id.
ValueError	If delete_source is True but the task properties are not properly set to identify the source to delete.

Delete a curation task asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask

syn = Synapse()
syn.login()

async def main():
    task = CurationTask(task_id=123)
    await task.delete_async()
    print("Task deleted successfully")

asyncio.run(main())

Delete a curation task and its associated data source asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask

syn = Synapse()
syn.login()

async def main():
    task = CurationTask(task_id=123)
    await task.delete_async(delete_source=True)
    print("Task and record set deleted successfully")

asyncio.run(main())

Source code in synapseclient/models/curation.py

async def delete_async(
    self,
    delete_source: bool = False,
    *,
    synapse_client: Optional[Synapse] = None,
) -> None:
    """
    Deletes a CurationTask from Synapse.

    Arguments:
        delete_source: If True, the associated source data (EntityView or RecordSet) will also be deleted
            if the task is a FileBasedMetadataTask or RecordBasedMetadataTask respectively. Defaults to False.
        synapse_client: If not passed in and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Raises:
        ValueError: If the CurationTask object does not have a task_id.
        ValueError: If delete_source is True but the task properties are not properly set
          to identify the source to delete.

    Example: Delete a curation task asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask

        syn = Synapse()
        syn.login()

        async def main():
            task = CurationTask(task_id=123)
            await task.delete_async()
            print("Task deleted successfully")

        asyncio.run(main())
        ```

    Example: Delete a curation task and its associated data source asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask

        syn = Synapse()
        syn.login()

        async def main():
            task = CurationTask(task_id=123)
            await task.delete_async(delete_source=True)
            print("Task and record set deleted successfully")

        asyncio.run(main())
        ```
    """
    if not self.task_id:
        raise ValueError("task_id is required to delete a CurationTask")

    trace.get_current_span().set_attributes(
        {
            "synapse.task_id": str(self.task_id),
        }
    )

    if delete_source:
        if not self.task_properties:
            await self.get_async(synapse_client=synapse_client)

        if isinstance(self.task_properties, FileBasedMetadataTaskProperties):
            if not self.task_properties.file_view_id:
                raise ValueError(
                    "Cannot delete Fileview: "
                    "'file_view_id' attribute is missing."
                )
            from synapseclient.models import EntityView

            await EntityView(id=self.task_properties.file_view_id).delete_async(
                synapse_client=synapse_client
            )

        elif isinstance(self.task_properties, RecordBasedMetadataTaskProperties):
            if not self.task_properties.record_set_id:
                raise ValueError(
                    "Cannot delete RecordSet: "
                    "'record_set_id' attribute is missing."
                )
            from synapseclient.models import RecordSet

            await RecordSet(id=self.task_properties.record_set_id).delete_async(
                synapse_client=synapse_client
            )

        else:
            raise ValueError(
                "'task_property' attribute is None. "
                "Deletion only supports FileBasedMetadataTaskProperties or "
                "RecordBasedMetadataTaskProperties."
            )

    await delete_curation_task(task_id=self.task_id, synapse_client=synapse_client)

store_async async

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

Creates a new CurationTask or updates an existing one on Synapse.

This method implements non-destructive updates. If a CurationTask with the same project_id and data_type exists and this instance hasn't been retrieved from Synapse before, it will merge the existing task data with the current instance before updating.

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
CurationTask	The CurationTask object. TYPE: CurationTask

Create a new curation task asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask, FileBasedMetadataTaskProperties

syn = Synapse()
syn.login()

async def main():
    # Create file-based task properties
    file_properties = FileBasedMetadataTaskProperties(
        upload_folder_id="syn1234567",
        file_view_id="syn2345678"
    )

    # Create and store the curation task
    task = CurationTask(
        project_id="syn9876543",
        data_type="genomics_data",
        instructions="Upload your genomics files to the specified folder",
        task_properties=file_properties
    )
    task = await task.store_async()
    print(f"Created task with ID: {task.task_id}")

asyncio.run(main())

Source code in synapseclient/models/curation.py

async def store_async(
    self, *, synapse_client: Optional[Synapse] = None
) -> "CurationTask":
    """
    Creates a new CurationTask or updates an existing one on Synapse.

    This method implements non-destructive updates. If a CurationTask with the same
    project_id and data_type exists and this instance hasn't been retrieved from
    Synapse before, it will merge the existing task data with the current instance
    before updating.

    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:
        CurationTask: The CurationTask object.

    Example: Create a new curation task asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask, FileBasedMetadataTaskProperties

        syn = Synapse()
        syn.login()

        async def main():
            # Create file-based task properties
            file_properties = FileBasedMetadataTaskProperties(
                upload_folder_id="syn1234567",
                file_view_id="syn2345678"
            )

            # Create and store the curation task
            task = CurationTask(
                project_id="syn9876543",
                data_type="genomics_data",
                instructions="Upload your genomics files to the specified folder",
                task_properties=file_properties
            )
            task = await task.store_async()
            print(f"Created task with ID: {task.task_id}")

        asyncio.run(main())
        ```
    """
    if not self.project_id:
        raise ValueError("project_id is required")
    if not self.data_type:
        raise ValueError("data_type is required")

    trace.get_current_span().set_attributes(
        {
            "synapse.data_type": self.data_type or "",
            "synapse.project_id": self.project_id or "",
            "synapse.task_id": str(self.task_id) if self.task_id else "",
        }
    )

    if (
        not self._last_persistent_instance
        and not self.task_id
        and (
            existing_task_id := await _get_existing_curation_task_id(
                project_id=self.project_id,
                data_type=self.data_type,
                synapse_client=synapse_client,
            )
        )
        and (
            existing_task := await CurationTask(task_id=existing_task_id).get_async(
                synapse_client=synapse_client
            )
        )
    ):
        merge_dataclass_entities(source=existing_task, destination=self)

    if self.task_id:
        task_result = await update_curation_task(
            task_id=self.task_id,
            curation_task=self.to_synapse_request(),
            synapse_client=synapse_client,
        )
        self.fill_from_dict(synapse_response=task_result)
        self._set_last_persistent_instance()
        return self
    else:
        if not self.project_id:
            raise ValueError("project_id is required to create a CurationTask")
        if not self.data_type:
            raise ValueError("data_type is required to create a CurationTask")
        if not self.instructions:
            raise ValueError("instructions is required to create a CurationTask")
        if not self.task_properties:
            raise ValueError("task_properties is required to create a CurationTask")

        task_result = await create_curation_task(
            curation_task=self.to_synapse_request(),
            synapse_client=synapse_client,
        )
        self.fill_from_dict(synapse_response=task_result)
        self._set_last_persistent_instance()
        return self

list_async async classmethod

list_async(project_id: str, *, assigned_to_me: Optional[bool] = None, assignee_ids: Optional[list[str]] = None, state_filter: Optional[list[Union[TaskState, str]]] = None, synapse_client: Optional[Synapse] = None) -> AsyncGenerator[CurationTask, None]

Generator that yields CurationTasks for a project as they become available.

PARAMETER	DESCRIPTION
project_id	The synId of the project. TYPE: str
assigned_to_me	When True, only return tasks assigned to the current user. Cannot be combined with assignee_ids. False does not mean "tasks not assigned to me". Defaults to None. TYPE: Optional[bool] DEFAULT: None
assignee_ids	Optional list of principal IDs (users or teams) to filter tasks by assignee. Cannot be combined with assigned_to_me=True. Passing an empty list raises a ValueError; pass None to return tasks for any assignee. Defaults to None. TYPE: Optional[list[str]] DEFAULT: None
state_filter	Optional list of TaskState values or exact-case strings to filter tasks by their current state (e.g., "IN_PROGRESS"). Defaults to None (all states returned). Passing an empty list raises a ValueError; pass None to return tasks in any state. TYPE: Optional[list[Union[TaskState, 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

YIELDS	DESCRIPTION
AsyncGenerator[CurationTask, None]	CurationTask objects as they are retrieved from the API.

RAISES	DESCRIPTION
ValueError	If state_filter is an empty list.
ValueError	If assignee_ids is an empty list.
ValueError	If assigned_to_me is True and assignee_ids is also provided.
ValueError	If any value in state_filter is not a TaskState member or an exact-case string matching a TaskState value (e.g., "IN_PROGRESS").
ValueError	If the Synapse response for any task does not contain taskProperties.

List all curation tasks in a project asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask

syn = Synapse()
syn.login()

async def main():
    # List all curation tasks in the project
    async for task in CurationTask.list_async(project_id="syn9876543"):
        print(f"Task ID: {task.task_id}")
        print(f"Data Type: {task.data_type}")
        print(f"Instructions: {task.instructions}")
        print("---")

asyncio.run(main())

List only curation tasks assigned to the current user asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask

syn = Synapse()
syn.login()

async def main():
    async for task in CurationTask.list_async(
        project_id="syn9876543", assigned_to_me=True
    ):
        print(f"Task ID: {task.task_id}")
        print(f"Data Type: {task.data_type}")
        print("---")

asyncio.run(main())

List only in-progress curation tasks asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask, TaskState

syn = Synapse()
syn.login()

async def main():
    async for task in CurationTask.list_async(
        project_id="syn9876543",
        state_filter=[TaskState.IN_PROGRESS],
    ):
        print(f"Task ID: {task.task_id}")
        print(f"Data Type: {task.data_type}")
        print("---")

asyncio.run(main())

List only in-progress curation tasks using a string state filter asynchronously

 

state_filter also accepts plain strings matching TaskState names exactly.

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask

syn = Synapse()
syn.login()

async def main():
    async for task in CurationTask.list_async(
        project_id="syn9876543",
        state_filter=["IN_PROGRESS"],
    ):
        print(f"Task ID: {task.task_id}")
        print(f"Data Type: {task.data_type}")
        print("---")

asyncio.run(main())

Source code in synapseclient/models/curation.py

@skip_async_to_sync
@classmethod
async def list_async(
    cls,
    project_id: str,
    *,
    assigned_to_me: Optional[bool] = None,
    assignee_ids: Optional[list[str]] = None,
    state_filter: Optional[list[Union["TaskState", str]]] = None,
    synapse_client: Optional[Synapse] = None,
) -> AsyncGenerator["CurationTask", None]:
    """
    Generator that yields CurationTasks for a project as they become available.

    Arguments:
        project_id: The synId of the project.
        assigned_to_me: When True, only return tasks assigned to the current user.
            Cannot be combined with assignee_ids.
            False does not mean "tasks not assigned to me".
            Defaults to None.
        assignee_ids: Optional list of principal IDs (users or teams) to filter
            tasks by assignee. Cannot be combined with assigned_to_me=True.
            Passing an empty list raises a ValueError; pass None to return tasks
            for any assignee. Defaults to None.
        state_filter: Optional list of TaskState values or exact-case strings to
            filter tasks by their current state (e.g., "IN_PROGRESS"). Defaults to
            None (all states returned). Passing an empty list raises a ValueError;
            pass None to return tasks in any state.
        synapse_client: If not passed in and caching was not disabled by
            Synapse.allow_client_caching(False) this will use the last created
            instance from the Synapse class constructor.

    Yields:
        CurationTask objects as they are retrieved from the API.

    Raises:
        ValueError: If state_filter is an empty list.
        ValueError: If assignee_ids is an empty list.
        ValueError: If assigned_to_me is True and assignee_ids is also provided.
        ValueError: If any value in state_filter is not a TaskState member or
            an exact-case string matching a TaskState value (e.g., "IN_PROGRESS").
        ValueError: If the Synapse response for any task does not contain
            taskProperties.

    Example: List all curation tasks in a project asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask

        syn = Synapse()
        syn.login()

        async def main():
            # List all curation tasks in the project
            async for task in CurationTask.list_async(project_id="syn9876543"):
                print(f"Task ID: {task.task_id}")
                print(f"Data Type: {task.data_type}")
                print(f"Instructions: {task.instructions}")
                print("---")

        asyncio.run(main())
        ```

    Example: List only curation tasks assigned to the current user asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask

        syn = Synapse()
        syn.login()

        async def main():
            async for task in CurationTask.list_async(
                project_id="syn9876543", assigned_to_me=True
            ):
                print(f"Task ID: {task.task_id}")
                print(f"Data Type: {task.data_type}")
                print("---")

        asyncio.run(main())
        ```

    Example: List only in-progress curation tasks asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask, TaskState

        syn = Synapse()
        syn.login()

        async def main():
            async for task in CurationTask.list_async(
                project_id="syn9876543",
                state_filter=[TaskState.IN_PROGRESS],
            ):
                print(f"Task ID: {task.task_id}")
                print(f"Data Type: {task.data_type}")
                print("---")

        asyncio.run(main())
        ```

    Example: List only in-progress curation tasks using a string state filter asynchronously
         

        state_filter also accepts plain strings matching TaskState names exactly.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask

        syn = Synapse()
        syn.login()

        async def main():
            async for task in CurationTask.list_async(
                project_id="syn9876543",
                state_filter=["IN_PROGRESS"],
            ):
                print(f"Task ID: {task.task_id}")
                print(f"Data Type: {task.data_type}")
                print("---")

        asyncio.run(main())
        ```
    """
    if state_filter == []:
        raise ValueError(
            "state_filter must not be empty. Pass None to return tasks in any state."
        )
    if assignee_ids == []:
        raise ValueError(
            "assignee_ids must not be empty. Pass None to return tasks for any assignee."
        )
    if assigned_to_me is True and assignee_ids is not None:
        raise ValueError(
            f"assigned_to_me and assignee_ids are mutually exclusive "
            f"and cannot be used together. Got assignee_ids={assignee_ids!r}."
        )

    if state_filter is not None:
        state_filter = coerce_enum_list(TaskState, state_filter)

    trace.get_current_span().set_attributes(
        {
            "synapse.project_id": project_id,
        }
    )

    async for task_dict in list_curation_tasks(
        project_id=project_id,
        assigned_to_me=assigned_to_me,
        assignee_ids=assignee_ids,
        state_filter=state_filter,
        synapse_client=synapse_client,
    ):
        task = cls().fill_from_dict(synapse_response=task_dict)
        yield task

create_grid_session_async async

create_grid_session_async(*, owner_principal_id: int | None = None, timeout: int = 120, synapse_client: Synapse | None = None) -> Grid

Create a new Grid session for this CurationTask and set it as the active session.

Picks the Grid seed from this task's task_properties:

• RecordBasedMetadataTaskProperties uses record_set_id
• FileBasedMetadataTaskProperties uses an initial_query that selects from the file_view_id

Always creates a new Grid session. To attach an existing session to a task, use set_active_grid_session_async instead.

After the Grid is created, updates the CurationTaskStatus to point its active_session_id at the new session. If that update fails for any reason, the newly created Grid is deleted on a best-effort basis and the original exception is re-raised.

PARAMETER	DESCRIPTION
owner_principal_id	The principal ID (user or team) that will own the created grid session. When not provided, the principal ID of the caller is used. TYPE: int | None DEFAULT: None
timeout	Seconds to wait for the grid creation job. Defaults to 120. TYPE: int DEFAULT: 120
synapse_client	If not passed 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: Synapse | None DEFAULT: None

RETURNS	DESCRIPTION
Grid	The newly created Grid.

RAISES	DESCRIPTION
ValueError	If task_id is unset or task_properties is of an unsupported type.
SynapseHTTPError	If the RecordSet or EntityView does not exist, or if the status update fails. The orphan Grid is deleted on a best-effort basis before the error is re-raised.

Create a grid session for a curation task asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask

syn = Synapse()
syn.login()

async def main():
    grid = await CurationTask(task_id=123).create_grid_session_async()
    print(grid.session_id)

asyncio.run(main())

Source code in synapseclient/models/curation.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: (
        f"CurationTask_CreateGridSession: ID: {self.task_id}"
    )
)
async def create_grid_session_async(
    self,
    *,
    owner_principal_id: int | None = None,
    timeout: int = 120,
    synapse_client: Synapse | None = None,
) -> "Grid":
    """
    Create a new Grid session for this CurationTask and set it as the active session.

    Picks the Grid seed from this task's task_properties:

    - RecordBasedMetadataTaskProperties uses record_set_id
    - FileBasedMetadataTaskProperties uses an initial_query that selects from
      the file_view_id

    Always creates a new Grid session. To attach an existing session to a task,
    use set_active_grid_session_async instead.

    After the Grid is created, updates the CurationTaskStatus to point its
    active_session_id at the new session. If that update fails for any reason,
    the newly created Grid is deleted on a best-effort basis and the original
    exception is re-raised.

    Arguments:
        owner_principal_id: The principal ID (user or team) that will own the
            created grid session. When not provided, the principal ID of the
            caller is used.
        timeout: Seconds to wait for the grid creation job. Defaults to 120.
        synapse_client: If not passed 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 newly created Grid.

    Raises:
        ValueError: If task_id is unset or task_properties is of an unsupported type.
        SynapseHTTPError: If the RecordSet or EntityView does not exist, or if the
            status update fails. The orphan Grid is deleted on a best-effort basis
            before the error is re-raised.

    Example: Create a grid session for a curation task asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask

        syn = Synapse()
        syn.login()

        async def main():
            grid = await CurationTask(task_id=123).create_grid_session_async()
            print(grid.session_id)

        asyncio.run(main())
        ```
    """
    if not self.task_id:
        raise ValueError(
            "task_id is required to create a CurationTask grid session"
        )

    if not self.task_properties:
        await self.get_async(synapse_client=synapse_client)

    if isinstance(self.task_properties, RecordBasedMetadataTaskProperties):
        if not self.task_properties.record_set_id:
            raise ValueError(
                "Cannot create grid session: "
                "task_properties.record_set_id is missing."
            )
        from synapseclient.models import RecordSet

        # raises SynapseHTTPError if RecordSet does not exist
        await RecordSet(id=self.task_properties.record_set_id).get_async(
            synapse_client=synapse_client
        )
        grid = Grid(
            record_set_id=self.task_properties.record_set_id,
            owner_principal_id=owner_principal_id,
        )
    elif isinstance(self.task_properties, FileBasedMetadataTaskProperties):
        if not self.task_properties.file_view_id:
            raise ValueError(
                "Cannot create grid session: "
                "task_properties.file_view_id is missing."
            )
        from synapseclient.models import EntityView

        # raises SynapseHTTPError if EntityView does not exist
        await EntityView(id=self.task_properties.file_view_id).get_async(
            synapse_client=synapse_client
        )
        grid = Grid(
            initial_query=Query(
                sql=f"SELECT * FROM {self.task_properties.file_view_id}"
            ),
            owner_principal_id=owner_principal_id,
        )
    else:
        raise ValueError(
            "task_properties must be a FileBasedMetadataTaskProperties or "
            "RecordBasedMetadataTaskProperties to create a grid session"
        )

    grid = await grid.create_async(
        timeout=timeout,
        synapse_client=synapse_client,
    )

    # Only one grid session can be set as the active one on a a given CurationTask
    # at a any time, though multiple sessions can exist.
    # If two users run this concurrently, one will lose the race and
    # receive a 412 (precondition failed). In that case — or if recording the
    # active session fails for any other reason — delete the session we just
    # created so it doesn't become an orphan. If the delete also fails, log a
    # warning so the caller knows manual cleanup is needed, then re-raise the
    # original exception in all cases.
    try:
        await self.set_active_grid_session_async(
            active_session_id=grid.session_id, synapse_client=synapse_client
        )
    except Exception:
        try:
            await grid.delete_async(synapse_client=synapse_client)
        except Exception:
            Synapse.get_client(synapse_client=synapse_client).logger.warning(
                "Failed to delete orphan grid session %s after status "
                "update failure; manual cleanup may be required.",
                grid.session_id,
            )
        raise

    return grid

set_task_state_async async

set_task_state_async(state: TaskState | str, *, synapse_client: Synapse | None = None) -> CurationTaskStatus

Set the state on this CurationTask's status.

Does not modify execution_details. Fetches the current CurationTaskStatus first so the update carries a fresh etag.

PARAMETER	DESCRIPTION
state	The state to set on this task's status. Accepts a TaskState or a string exactly matching one of its members (e.g. NOT_STARTED, IN_PROGRESS, COMPLETED, CANCELED). TYPE: TaskState | 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: Synapse | None DEFAULT: None

RETURNS	DESCRIPTION
CurationTaskStatus	The updated CurationTaskStatus object.

RAISES	DESCRIPTION
ValueError	If the CurationTask object does not have a task_id, or if state is a string that does not match a TaskState member.

Mark a curation task as completed asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask, TaskState

syn = Synapse()
syn.login()

async def main():
    await CurationTask(task_id=123).set_task_state_async(
        state=TaskState.COMPLETED
    )

asyncio.run(main())

Mark a curation task as completed using a string asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import CurationTask

syn = Synapse()
syn.login()

async def main():
    await CurationTask(task_id=123).set_task_state_async(
        state="COMPLETED"
    )

asyncio.run(main())

Source code in synapseclient/models/curation.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: (
        f"CurationTask_SetTaskState: ID: {self.task_id}"
    )
)
async def set_task_state_async(
    self,
    state: "TaskState | str",
    *,
    synapse_client: Synapse | None = None,
) -> "CurationTaskStatus":
    """
    Set the state on this CurationTask's status.

    Does not modify execution_details. Fetches the current CurationTaskStatus
    first so the update carries a fresh etag.

    Arguments:
        state: The state to set on this task's status. Accepts a
            TaskState or a string exactly matching one of its members
            (e.g. NOT_STARTED, IN_PROGRESS, COMPLETED, CANCELED).
        synapse_client: If not passed in and caching was not disabled by
            Synapse.allow_client_caching(False) this will use the last created
            instance from the Synapse class constructor.

    Returns:
        The updated CurationTaskStatus object.

    Raises:
        ValueError: If the CurationTask object does not have a task_id, or
            if state is a string that does not match a TaskState member.

    Example: Mark a curation task as completed asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask, TaskState

        syn = Synapse()
        syn.login()

        async def main():
            await CurationTask(task_id=123).set_task_state_async(
                state=TaskState.COMPLETED
            )

        asyncio.run(main())
        ```

    Example: Mark a curation task as completed using a string asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import CurationTask

        syn = Synapse()
        syn.login()

        async def main():
            await CurationTask(task_id=123).set_task_state_async(
                state="COMPLETED"
            )

        asyncio.run(main())
        ```
    """
    try:
        coerced_state = TaskState(state)
    except ValueError as exc:
        raise ValueError(
            f"{state!r} is not a valid TaskState. "
            f"Expected one of: {[s.value for s in TaskState]}."
        ) from exc

    status = await self.get_status_async(synapse_client=synapse_client)
    status.state = coerced_state
    return await self.update_status_async(
        curation_task_status=status, synapse_client=synapse_client
    )

---

synapseclient.models.RecordSet dataclass

Bases: RecordSetSynchronousProtocol, AccessControllable, BaseJSONSchema

A RecordSet entity captures record-based metadata as a special type of CSV. The record set content can be curated using the grid services. When a grid is created from a record set, its data can be exported back to a new version of the record set. The export will include the validation summary as well as a validation file handle that contains detailed validation results for each row in the record set.

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 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. 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 entity. The key is the name of your desired annotations. The value is an object containing a list of values (use empty list to represent no values for key) and the value type associated with all values in the list. To remove all annotations set this to an empty dict {}. TYPE: Optional[Dict[str, Union[List[str], List[bool], List[float], List[int], List[date], List[datetime]]]]
upsert_keys	One or more column names that define this upsert key for this set. This key is used to determine if a new record should be treated as an update or an insert. TYPE: Optional[List[str]]
csv_descriptor	The description of a CSV for upload or download. TYPE: Optional[CsvTableDescriptor]
validation_summary	Summary statistics for the JSON schema validation results for the children of an Entity container (Project or Folder). TYPE: Optional[ValidationSummary]
file_name_override	An optional replacement for the name of the uploaded file. This is distinct from the entity name. If omitted the file will retain its original name. TYPE: Optional[str]
validation_file_handle_id	(Read Only) Pointer to a CSV file that contains the detailed validation results for each row in the record set. The CSV file will contain for each row the following columns: row_index, is_valid, validation_error_message, all_validation_messages. Generated only from a grid session export, cannot be changed by the user. TYPE: Optional[str]
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. 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]
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. 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. TYPE: bool
synapse_store	(Store only) Whether the File should be uploaded or if false: only the path should be stored when synapseclient.models.File.store is called. TYPE: bool
download_file	(Get only) If True the file will be downloaded. TYPE: bool
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]
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/recordset.py

@dataclass()
@async_to_sync
class RecordSet(RecordSetSynchronousProtocol, AccessControllable, BaseJSONSchema):
    """
    A RecordSet entity captures record-based metadata as a special type of CSV.
    The record set content can be curated using the grid services.
    When a grid is created from a record set, its data can be exported back to a new version of the record set.
    The export will include the validation summary as well as a validation file handle that
      contains detailed validation results for each row in the record set.

    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 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.
        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 entity. 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 `{}`.
        upsert_keys: One or more column names that define this upsert key for this
            set. This key is used to determine if a new record should be treated as
            an update or an insert.
        csv_descriptor: The description of a CSV for upload or download.
        validation_summary: Summary statistics for the JSON schema validation results
            for the children of an Entity container (Project or Folder).
        file_name_override: An optional replacement for the name of the uploaded
            file. This is distinct from the entity name. If omitted the file will
            retain its original name.
        validation_file_handle_id: (Read Only) Pointer to a CSV file that contains the
            detailed validation results for each row in the record set. The CSV file
            will contain for each row the following columns: row_index, is_valid,
            validation_error_message, all_validation_messages. Generated only from a
            grid session export, cannot be changed by the user.
        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. 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.
        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.

            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.
        synapse_store: (Store only) Whether the File should be uploaded or if false:
            only the path should be stored when [synapseclient.models.File.store][]
            is called.
        download_file: (Get only) If True the file will be downloaded.
        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.
        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.
    """

    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 `{}`."""

    upsert_keys: Optional[List[str]] = field(default_factory=list)
    """One or more column names that define this upsert key for this set. This key is
    used to determine if a new record should be treated as an update or an insert.
    """

    csv_descriptor: Optional["CsvTableDescriptor"] = field(default=None)
    """The description of a CSV for upload or download."""

    validation_summary: Optional[ValidationSummary] = field(default=None, compare=False)
    """Summary statistics for the JSON schema validation results for the children of
    an Entity container (Project or Folder)"""

    file_name_override: Optional[str] = None
    """An optional replacement for the name of the uploaded file. This is distinct from
    the entity name. If omitted the file will retain its original name.
    """

    validation_file_handle_id: Optional[str] = None
    """
    (Read Only) Pointer to a CSV file that contains the detailed validation results for
    each row in the record set. The CSV file will contain for each row the following
    columns: row_index, is_valid, validation_error_message, all_validation_messages.
    Generated only from a grid session export, cannot be changed by the user.
    """

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

    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["RecordSet"] = 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

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

        This method populates the RecordSet instance with data from a Synapse REST API
        response or a dictionary containing RecordSet information. It handles the
        conversion from Synapse API field names (camelCase) to Python attribute names
        (snake_case) and processes nested objects appropriately.

        Arguments:
            synapse_file: The response from the REST API or a dictionary containing
                RecordSet data. Can be either a Synapse_File object or a dictionary
                with string keys and various value types.
            set_annotations: Whether to set the annotations from the response.
                If True, annotations will be populated from the API response.

        Returns:
            The RecordSet object with updated attributes from the API response.
        """
        self.id = entity.get("id", None)
        self.name = entity.get("name", None)
        self.description = entity.get("description", None)
        self.etag = entity.get("etag", None)
        self.created_on = entity.get("createdOn", None)
        self.modified_on = entity.get("modifiedOn", None)
        self.created_by = entity.get("createdBy", None)
        self.modified_by = entity.get("modifiedBy", None)
        self.parent_id = entity.get("parentId", None)
        self.version_number = entity.get("versionNumber", None)
        self.version_label = entity.get("versionLabel", None)
        self.version_comment = entity.get("versionComment", None)
        self.is_latest_version = entity.get("isLatestVersion", False)
        self.data_file_handle_id = entity.get("dataFileHandleId", None)
        self.path = entity.get("path", self.path)
        self.file_name_override = entity.get("fileNameOverride", None)
        self.validation_file_handle_id = entity.get("validationFileHandleId", None)
        csv_descriptor = entity.get("csvDescriptor", None)
        if csv_descriptor:
            from synapseclient.models import CsvTableDescriptor

            self.csv_descriptor = CsvTableDescriptor().fill_from_dict(csv_descriptor)

        validation_summary = entity.get("validationSummary", None)

        if validation_summary:
            self.validation_summary = ValidationSummary(
                container_id=validation_summary.get("containerId", None),
                total_number_of_children=validation_summary.get(
                    "totalNumberOfChildren", None
                ),
                number_of_valid_children=validation_summary.get(
                    "numberOfValidChildren", None
                ),
                number_of_invalid_children=validation_summary.get(
                    "numberOfInvalidChildren", None
                ),
                number_of_unknown_children=validation_summary.get(
                    "numberOfUnknownChildren", None
                ),
                generated_on=validation_summary.get("generatedOn", None),
            )

        self.upsert_keys = entity.get("upsertKey", [])

        synapse_file_handle = entity.get("_file_handle", None)
        if synapse_file_handle:
            from synapseclient.models import FileHandle

            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(entity.get("annotations", {}))
        return self

    def _cannot_store(self) -> bool:
        """
        Determines based on guard conditions if a store operation can proceed.
        """
        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 hash of a local file if it exists and hasn't been loaded yet.

        This method computes and sets the content_md5 attribute for local files
        that exist on disk. It only performs the calculation if the content_md5
        is not already set and the path points to an existing file.
        """
        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_entity(
        self, *, synapse_client: Optional[Synapse] = None
    ) -> Union["RecordSet", None]:
        """
        Determines if the RecordSet already exists in Synapse.

        This method searches for an existing RecordSet in Synapse that matches the
        current instance. If found, it returns the existing RecordSet object, otherwise
        it returns None. This is used to determine if the RecordSet should be updated
        or created during a store operation.

        Arguments:
            synapse_client: If not passed in and caching was not disabled, this will
                use the last created instance from the Synapse class constructor.

        Returns:
            The existing RecordSet object if it exists in Synapse, None otherwise.
        """

        async def get_entity(existing_id: str) -> "RecordSet":
            """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:
                entity_copy = RecordSet(
                    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 entity_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_entity_id := await get_id(
                    entity=self,
                    failure_strategy=None,
                    synapse_client=synapse_client,
                )
            )
            and (existing_file := await get_entity(existing_entity_id))
        ):
            return existing_file
        return None

    def _determine_fields_to_ignore_in_merge(self) -> List[str]:
        """
        Determine which fields should not be merged when merging two entities.

        This method returns a list of field names that should be ignored during
        entity merging operations. This allows for fine-tuned destructive updates
        of an entity based on the current configuration settings.

        The method has special handling for manifest uploads where specific fields
        are provided in the manifest and should take precedence over existing
        entity values.

        Returns:
            A list of field names that should not be merged from the existing 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")

        return fields_to_not_merge

    def to_synapse_request(self) -> Dict[str, Any]:
        """
        Converts this dataclass to a dictionary suitable for a Synapse REST API request.

        This method transforms the RecordSet object into a dictionary format that
        matches the structure expected by the Synapse REST API. It handles the
        conversion of Python snake_case attribute names to the camelCase format
        used by the API, and ensures that nested objects are properly serialized.

        Returns:
            A dictionary representation of this object formatted for API requests.
            None values are automatically removed from the dictionary.

        Example: Converting a RecordSet for API submission
            This method is used internally when storing or updating RecordSets:

            ```python
            from synapseclient.models import RecordSet

            record_set = RecordSet(
                name="My RecordSet",
                description="A test record set",
                parent_id="syn123456"
            )
            api_dict = record_set.to_synapse_request()
            # api_dict contains properly formatted data for the REST API
            ```
        """

        entity = {
            "concreteType": concrete_types.RECORD_SET_ENTITY,
            "name": self.name,
            "description": self.description,
            "id": self.id,
            "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,
            "isLatestVersion": self.is_latest_version,
            "dataFileHandleId": self.data_file_handle_id,
            "upsertKey": self.upsert_keys,
            "csvDescriptor": (
                self.csv_descriptor.to_synapse_request()
                if self.csv_descriptor
                else None
            ),
            "validationSummary": (
                {
                    "containerId": self.validation_summary.container_id,
                    "totalNumberOfChildren": self.validation_summary.total_number_of_children,
                    "numberOfValidChildren": self.validation_summary.number_of_valid_children,
                    "numberOfInvalidChildren": self.validation_summary.number_of_invalid_children,
                    "numberOfUnknownChildren": self.validation_summary.number_of_unknown_children,
                    "generatedOn": self.validation_summary.generated_on,
                }
                if self.validation_summary
                else None
            ),
            "fileNameOverride": self.file_name_override,
        }
        delete_none_keys(entity)

        return entity

    async def store_async(
        self,
        parent: Optional[Union["Folder", "Project"]] = None,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> "RecordSet":
        """
        Store the RecordSet in Synapse.

        This method uploads or updates a RecordSet in Synapse. It can handle both
        creating new RecordSets and updating existing ones based on the
        `create_or_update` flag. The method supports file uploads, metadata updates,
        and merging with existing entities when appropriate.

        Arguments:
            parent: The parent Folder or Project for this RecordSet. If provided,
                this will override the `parent_id` attribute.
            synapse_client: If not passed 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 RecordSet object with updated metadata from Synapse after the
            store operation.

        Raises:
            ValueError: If the RecordSet does not have the required information
                for storing. Must have either: (ID with path or data_file_handle_id),
                or (path with parent_id), or (data_file_handle_id with parent_id).

        Example: Storing a new RecordSet
            Creating and storing a new RecordSet in Synapse:

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import RecordSet

            async def main():
                syn = Synapse()
                syn.login()

                record_set = RecordSet(
                    name="My RecordSet",
                    description="A dataset for analysis",
                    parent_id="syn123456",
                    path="/path/to/data.csv"
                )
                stored_record_set = await record_set.store_async()
                print(f"Stored RecordSet with ID: {stored_record_set.id}")

            asyncio.run(main())
            ```

            Updating an existing RecordSet:
            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import RecordSet

            async def main():
                syn = Synapse()
                syn.login()

                record_set = await RecordSet(id="syn789012").get_async()
                record_set.description = "Updated description"
                updated_record_set = await record_set.store_async()

            asyncio.run(main())
            ```
        """
        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_entity(synapse_client=client):
            merge_dataclass_entities(
                source=existing_file,
                destination=self,
                fields_to_ignore=self._determine_fields_to_ignore_in_merge(),
            )

        if self.id:
            trace.get_current_span().set_attributes(
                {
                    "synapse.id": self.id,
                }
            )

        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()
            ):
                from synapseclient.models.file import _upload_file

                await _upload_file(entity_to_upload=self, 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:
            entity = await store_entity(
                resource=self, entity=self.to_synapse_request(), synapse_client=client
            )

            self.fill_from_dict(entity=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

    async def get_async(
        self,
        include_activity: bool = False,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> "RecordSet":
        """
        Get the RecordSet from Synapse.

        This method retrieves a RecordSet entity from Synapse. You may retrieve
        a RecordSet by either its ID or path. If you specify both, the ID will
        take precedence.

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

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

        Arguments:
            include_activity: If True, the activity will be included in the RecordSet
                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 RecordSet object with data populated from Synapse.

        Raises:
            ValueError: If the RecordSet does not have an ID or path to retrieve.

        Example: Retrieving a RecordSet by ID
            Get an existing RecordSet from Synapse:

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import RecordSet

            async def main():
                syn = Synapse()
                syn.login()

                record_set = await RecordSet(id="syn123").get_async()
                print(f"RecordSet name: {record_set.name}")

            asyncio.run(main())
            ```

            Downloading a RecordSet to a specific directory:
            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import RecordSet

            async def main():
                syn = Synapse()
                syn.login()

                record_set = await RecordSet(
                    id="syn123",
                    path="/path/to/download/directory"
                ).get_async()

            asyncio.run(main())
            ```

            Including activity information:
            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import RecordSet

            async def main():
                syn = Synapse()
                syn.login()

                record_set = await RecordSet(id="syn123").get_async(include_activity=True)
                if record_set.activity:
                    print(f"Activity: {record_set.activity.name}")

            asyncio.run(main())
            ```
        """
        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

    async def get_detailed_validation_results_async(
        self,
        download_location: Optional[str] = None,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> DATA_FRAME_TYPE:
        """
        Get detailed validation results for the RecordSet as a pandas DataFrame.

        This method downloads a CSV file containing detailed validation results for each row
        in the RecordSet. The validation results are generated when a RecordSet with a bound
        JSON schema is exported from a Grid session. The CSV contains columns:
        - row_index: The index of the row in the RecordSet
        - is_valid: Boolean indicating if the row is valid according to the schema
        - validation_error_message: The primary validation error message (if any)
        - all_validation_messages: All validation messages for the row (if any)

        Arguments:
            download_location: Optional directory path where the validation results CSV
                should be downloaded. If not specified, the file will be downloaded to
                the Synapse cache directory. If the file is already cached, it will use
                the cached version unless a different download_location is specified.
            synapse_client: If not passed 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 pandas DataFrame containing the validation results, or None if no
            validation_file_handle_id is available (with a warning logged).

        Example: Get validation results for a RecordSet
            Get detailed validation results after exporting from a Grid session:

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import RecordSet, Grid

            async def main():
                syn = Synapse()
                syn.login()

                # Assuming you have a RecordSet with a bound schema
                record_set = await RecordSet(id="syn123").get_async()

                # Create and export Grid session to generate validation results
                grid = await Grid(record_set_id=record_set.id).create_async()
                await grid.export_to_record_set_async()
                await grid.delete_async()

                # Re-fetch the RecordSet to get updated validation_file_handle_id
                record_set = await record_set.get_async()

                # Get the detailed validation results
                results_df = await record_set.get_detailed_validation_results_async()

                # Analyze the results
                print(f"Total rows: {len(results_df)}")
                print(f"Columns: {results_df.columns.tolist()}")

                # Filter for valid and invalid rows
                # Note: is_valid is boolean (True/False) for validated rows
                valid_rows = results_df[results_df['is_valid'] == True]  # noqa: E712
                invalid_rows = results_df[results_df['is_valid'] == False]  # noqa: E712

                print(f"Valid rows: {len(valid_rows)}")
                print(f"Invalid rows: {len(invalid_rows)}")

                # View invalid rows with their error messages
                if len(invalid_rows) > 0:
                    print(invalid_rows[['row_index', 'validation_error_message']])

            asyncio.run(main())
            ```
        """
        test_import_pandas()
        import pandas as pd

        client = Synapse.get_client(synapse_client=synapse_client)

        if not self.validation_file_handle_id:
            client.logger.warning(
                "No validation file handle ID found for this RecordSet. Cannot retrieve detailed validation results."
            )
            return None

        cached_file_path = client.cache.get(
            file_handle_id=self.validation_file_handle_id, path=download_location
        )

        # location in .synapseCache where the file would be corresponding to its FileHandleId
        synapse_cache_location = client.cache.get_cache_dir(
            file_handle_id=self.validation_file_handle_id
        )

        if download_location is not None:
            # Make sure the specified download location is a fully resolved directory
            download_location = ensure_download_location_is_directory(download_location)
        elif cached_file_path is not None:
            # file already cached so use that as the download location
            download_location = os.path.dirname(cached_file_path)
        else:
            # file not cached and no user-specified location so default to .synapseCache
            download_location = synapse_cache_location

        # Generate filename for the validation results CSV
        filename = f"SYNAPSE_RECORDSET_VALIDATION_{self.validation_file_handle_id}.csv"
        destination_path = os.path.join(download_location, filename)

        validation_file_path = await download_by_file_handle(
            file_handle_id=self.validation_file_handle_id,
            synapse_id=self.id,
            entity_type="FileEntity",
            destination=destination_path,
            synapse_client=client,
        )

        return pd.read_csv(validation_file_path)

    async def delete_async(
        self,
        version_only: Optional[bool] = False,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> None:
        """
        Delete the RecordSet from Synapse using its ID.

        This method removes a RecordSet entity from Synapse. You can choose to
        delete either a specific version or the entire RecordSet including all
        its versions.

        Arguments:
            version_only: If True, only the version specified in the `version_number`
                attribute of the RecordSet will be deleted. If False, the entire
                RecordSet including all versions 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 RecordSet does not have an ID to delete.
            ValueError: If the RecordSet does not have a version number to delete a
                specific version, and `version_only` is True.

        Example: Deleting a RecordSet
            Delete an entire RecordSet and all its versions:

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import RecordSet

            async def main():
                syn = Synapse()
                syn.login()

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

            asyncio.run(main())
            ```

            Delete only a specific version:
            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import RecordSet

            async def main():
                syn = Synapse()
                syn.login()

                record_set = RecordSet(id="syn123", version_number=2)
                await record_set.delete_async(version_only=True)

            asyncio.run(main())
            ```
        """
        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}"
        )

Methods:

get_async async

get_async(include_activity: bool = False, *, synapse_client: Optional[Synapse] = None) -> RecordSet

Get the RecordSet from Synapse.

This method retrieves a RecordSet entity from Synapse. You may retrieve a RecordSet by either its ID or path. If you specify both, the ID will take precedence.

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

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

PARAMETER	DESCRIPTION
include_activity	If True, the activity will be included in the RecordSet 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
RecordSet	The RecordSet object with data populated from Synapse.

RAISES	DESCRIPTION
ValueError	If the RecordSet does not have an ID or path to retrieve.

Retrieving a RecordSet by ID

Get an existing RecordSet from Synapse:

import asyncio
from synapseclient import Synapse
from synapseclient.models import RecordSet

async def main():
    syn = Synapse()
    syn.login()

    record_set = await RecordSet(id="syn123").get_async()
    print(f"RecordSet name: {record_set.name}")

asyncio.run(main())

Downloading a RecordSet to a specific directory:

import asyncio
from synapseclient import Synapse
from synapseclient.models import RecordSet

async def main():
    syn = Synapse()
    syn.login()

    record_set = await RecordSet(
        id="syn123",
        path="/path/to/download/directory"
    ).get_async()

asyncio.run(main())

Including activity information:

import asyncio
from synapseclient import Synapse
from synapseclient.models import RecordSet

async def main():
    syn = Synapse()
    syn.login()

    record_set = await RecordSet(id="syn123").get_async(include_activity=True)
    if record_set.activity:
        print(f"Activity: {record_set.activity.name}")

asyncio.run(main())

Source code in synapseclient/models/recordset.py

async def get_async(
    self,
    include_activity: bool = False,
    *,
    synapse_client: Optional[Synapse] = None,
) -> "RecordSet":
    """
    Get the RecordSet from Synapse.

    This method retrieves a RecordSet entity from Synapse. You may retrieve
    a RecordSet by either its ID or path. If you specify both, the ID will
    take precedence.

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

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

    Arguments:
        include_activity: If True, the activity will be included in the RecordSet
            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 RecordSet object with data populated from Synapse.

    Raises:
        ValueError: If the RecordSet does not have an ID or path to retrieve.

    Example: Retrieving a RecordSet by ID
        Get an existing RecordSet from Synapse:

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import RecordSet

        async def main():
            syn = Synapse()
            syn.login()

            record_set = await RecordSet(id="syn123").get_async()
            print(f"RecordSet name: {record_set.name}")

        asyncio.run(main())
        ```

        Downloading a RecordSet to a specific directory:
        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import RecordSet

        async def main():
            syn = Synapse()
            syn.login()

            record_set = await RecordSet(
                id="syn123",
                path="/path/to/download/directory"
            ).get_async()

        asyncio.run(main())
        ```

        Including activity information:
        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import RecordSet

        async def main():
            syn = Synapse()
            syn.login()

            record_set = await RecordSet(id="syn123").get_async(include_activity=True)
            if record_set.activity:
                print(f"Activity: {record_set.activity.name}")

        asyncio.run(main())
        ```
    """
    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) -> RecordSet

Store the RecordSet in Synapse.

This method uploads or updates a RecordSet in Synapse. It can handle both creating new RecordSets and updating existing ones based on the create_or_update flag. The method supports file uploads, metadata updates, and merging with existing entities when appropriate.

PARAMETER	DESCRIPTION
parent	The parent Folder or Project for this RecordSet. If provided, this will override the parent_id attribute. 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
RecordSet	The RecordSet object with updated metadata from Synapse after the
RecordSet	store operation.

RAISES	DESCRIPTION
ValueError	If the RecordSet does not have the required information for storing. Must have either: (ID with path or data_file_handle_id), or (path with parent_id), or (data_file_handle_id with parent_id).

Storing a new RecordSet

Creating and storing a new RecordSet in Synapse:

import asyncio
from synapseclient import Synapse
from synapseclient.models import RecordSet

async def main():
    syn = Synapse()
    syn.login()

    record_set = RecordSet(
        name="My RecordSet",
        description="A dataset for analysis",
        parent_id="syn123456",
        path="/path/to/data.csv"
    )
    stored_record_set = await record_set.store_async()
    print(f"Stored RecordSet with ID: {stored_record_set.id}")

asyncio.run(main())

Updating an existing RecordSet:

import asyncio
from synapseclient import Synapse
from synapseclient.models import RecordSet

async def main():
    syn = Synapse()
    syn.login()

    record_set = await RecordSet(id="syn789012").get_async()
    record_set.description = "Updated description"
    updated_record_set = await record_set.store_async()

asyncio.run(main())

Source code in synapseclient/models/recordset.py

async def store_async(
    self,
    parent: Optional[Union["Folder", "Project"]] = None,
    *,
    synapse_client: Optional[Synapse] = None,
) -> "RecordSet":
    """
    Store the RecordSet in Synapse.

    This method uploads or updates a RecordSet in Synapse. It can handle both
    creating new RecordSets and updating existing ones based on the
    `create_or_update` flag. The method supports file uploads, metadata updates,
    and merging with existing entities when appropriate.

    Arguments:
        parent: The parent Folder or Project for this RecordSet. If provided,
            this will override the `parent_id` attribute.
        synapse_client: If not passed 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 RecordSet object with updated metadata from Synapse after the
        store operation.

    Raises:
        ValueError: If the RecordSet does not have the required information
            for storing. Must have either: (ID with path or data_file_handle_id),
            or (path with parent_id), or (data_file_handle_id with parent_id).

    Example: Storing a new RecordSet
        Creating and storing a new RecordSet in Synapse:

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import RecordSet

        async def main():
            syn = Synapse()
            syn.login()

            record_set = RecordSet(
                name="My RecordSet",
                description="A dataset for analysis",
                parent_id="syn123456",
                path="/path/to/data.csv"
            )
            stored_record_set = await record_set.store_async()
            print(f"Stored RecordSet with ID: {stored_record_set.id}")

        asyncio.run(main())
        ```

        Updating an existing RecordSet:
        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import RecordSet

        async def main():
            syn = Synapse()
            syn.login()

            record_set = await RecordSet(id="syn789012").get_async()
            record_set.description = "Updated description"
            updated_record_set = await record_set.store_async()

        asyncio.run(main())
        ```
    """
    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_entity(synapse_client=client):
        merge_dataclass_entities(
            source=existing_file,
            destination=self,
            fields_to_ignore=self._determine_fields_to_ignore_in_merge(),
        )

    if self.id:
        trace.get_current_span().set_attributes(
            {
                "synapse.id": self.id,
            }
        )

    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()
        ):
            from synapseclient.models.file import _upload_file

            await _upload_file(entity_to_upload=self, 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:
        entity = await store_entity(
            resource=self, entity=self.to_synapse_request(), synapse_client=client
        )

        self.fill_from_dict(entity=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

delete_async async

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

Delete the RecordSet from Synapse using its ID.

This method removes a RecordSet entity from Synapse. You can choose to delete either a specific version or the entire RecordSet including all its versions.

PARAMETER	DESCRIPTION
version_only	If True, only the version specified in the version_number attribute of the RecordSet will be deleted. If False, the entire RecordSet including all versions 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 RecordSet does not have an ID to delete.
ValueError	If the RecordSet does not have a version number to delete a specific version, and version_only is True.

Deleting a RecordSet

Delete an entire RecordSet and all its versions:

import asyncio
from synapseclient import Synapse
from synapseclient.models import RecordSet

async def main():
    syn = Synapse()
    syn.login()

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

asyncio.run(main())

Delete only a specific version:

import asyncio
from synapseclient import Synapse
from synapseclient.models import RecordSet

async def main():
    syn = Synapse()
    syn.login()

    record_set = RecordSet(id="syn123", version_number=2)
    await record_set.delete_async(version_only=True)

asyncio.run(main())

Source code in synapseclient/models/recordset.py

async def delete_async(
    self,
    version_only: Optional[bool] = False,
    *,
    synapse_client: Optional[Synapse] = None,
) -> None:
    """
    Delete the RecordSet from Synapse using its ID.

    This method removes a RecordSet entity from Synapse. You can choose to
    delete either a specific version or the entire RecordSet including all
    its versions.

    Arguments:
        version_only: If True, only the version specified in the `version_number`
            attribute of the RecordSet will be deleted. If False, the entire
            RecordSet including all versions 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 RecordSet does not have an ID to delete.
        ValueError: If the RecordSet does not have a version number to delete a
            specific version, and `version_only` is True.

    Example: Deleting a RecordSet
        Delete an entire RecordSet and all its versions:

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import RecordSet

        async def main():
            syn = Synapse()
            syn.login()

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

        asyncio.run(main())
        ```

        Delete only a specific version:
        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import RecordSet

        async def main():
            syn = Synapse()
            syn.login()

            record_set = RecordSet(id="syn123", version_number=2)
            await record_set.delete_async(version_only=True)

        asyncio.run(main())
        ```
    """
    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}"
    )

get_detailed_validation_results_async async

get_detailed_validation_results_async(download_location: Optional[str] = None, *, synapse_client: Optional[Synapse] = None) -> DataFrame

Get detailed validation results for the RecordSet as a pandas DataFrame.

This method downloads a CSV file containing detailed validation results for each row in the RecordSet. The validation results are generated when a RecordSet with a bound JSON schema is exported from a Grid session. The CSV contains columns: - row_index: The index of the row in the RecordSet - is_valid: Boolean indicating if the row is valid according to the schema - validation_error_message: The primary validation error message (if any) - all_validation_messages: All validation messages for the row (if any)

PARAMETER	DESCRIPTION
download_location	Optional directory path where the validation results CSV should be downloaded. If not specified, the file will be downloaded to the Synapse cache directory. If the file is already cached, it will use the cached version unless a different download_location is specified. 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
DataFrame	A pandas DataFrame containing the validation results, or None if no
DataFrame	validation_file_handle_id is available (with a warning logged).

Get validation results for a RecordSet

Get detailed validation results after exporting from a Grid session:

import asyncio
from synapseclient import Synapse
from synapseclient.models import RecordSet, Grid

async def main():
    syn = Synapse()
    syn.login()

    # Assuming you have a RecordSet with a bound schema
    record_set = await RecordSet(id="syn123").get_async()

    # Create and export Grid session to generate validation results
    grid = await Grid(record_set_id=record_set.id).create_async()
    await grid.export_to_record_set_async()
    await grid.delete_async()

    # Re-fetch the RecordSet to get updated validation_file_handle_id
    record_set = await record_set.get_async()

    # Get the detailed validation results
    results_df = await record_set.get_detailed_validation_results_async()

    # Analyze the results
    print(f"Total rows: {len(results_df)}")
    print(f"Columns: {results_df.columns.tolist()}")

    # Filter for valid and invalid rows
    # Note: is_valid is boolean (True/False) for validated rows
    valid_rows = results_df[results_df['is_valid'] == True]  # noqa: E712
    invalid_rows = results_df[results_df['is_valid'] == False]  # noqa: E712

    print(f"Valid rows: {len(valid_rows)}")
    print(f"Invalid rows: {len(invalid_rows)}")

    # View invalid rows with their error messages
    if len(invalid_rows) > 0:
        print(invalid_rows[['row_index', 'validation_error_message']])

asyncio.run(main())

Source code in synapseclient/models/recordset.py

async def get_detailed_validation_results_async(
    self,
    download_location: Optional[str] = None,
    *,
    synapse_client: Optional[Synapse] = None,
) -> DATA_FRAME_TYPE:
    """
    Get detailed validation results for the RecordSet as a pandas DataFrame.

    This method downloads a CSV file containing detailed validation results for each row
    in the RecordSet. The validation results are generated when a RecordSet with a bound
    JSON schema is exported from a Grid session. The CSV contains columns:
    - row_index: The index of the row in the RecordSet
    - is_valid: Boolean indicating if the row is valid according to the schema
    - validation_error_message: The primary validation error message (if any)
    - all_validation_messages: All validation messages for the row (if any)

    Arguments:
        download_location: Optional directory path where the validation results CSV
            should be downloaded. If not specified, the file will be downloaded to
            the Synapse cache directory. If the file is already cached, it will use
            the cached version unless a different download_location is specified.
        synapse_client: If not passed 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 pandas DataFrame containing the validation results, or None if no
        validation_file_handle_id is available (with a warning logged).

    Example: Get validation results for a RecordSet
        Get detailed validation results after exporting from a Grid session:

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import RecordSet, Grid

        async def main():
            syn = Synapse()
            syn.login()

            # Assuming you have a RecordSet with a bound schema
            record_set = await RecordSet(id="syn123").get_async()

            # Create and export Grid session to generate validation results
            grid = await Grid(record_set_id=record_set.id).create_async()
            await grid.export_to_record_set_async()
            await grid.delete_async()

            # Re-fetch the RecordSet to get updated validation_file_handle_id
            record_set = await record_set.get_async()

            # Get the detailed validation results
            results_df = await record_set.get_detailed_validation_results_async()

            # Analyze the results
            print(f"Total rows: {len(results_df)}")
            print(f"Columns: {results_df.columns.tolist()}")

            # Filter for valid and invalid rows
            # Note: is_valid is boolean (True/False) for validated rows
            valid_rows = results_df[results_df['is_valid'] == True]  # noqa: E712
            invalid_rows = results_df[results_df['is_valid'] == False]  # noqa: E712

            print(f"Valid rows: {len(valid_rows)}")
            print(f"Invalid rows: {len(invalid_rows)}")

            # View invalid rows with their error messages
            if len(invalid_rows) > 0:
                print(invalid_rows[['row_index', 'validation_error_message']])

        asyncio.run(main())
        ```
    """
    test_import_pandas()
    import pandas as pd

    client = Synapse.get_client(synapse_client=synapse_client)

    if not self.validation_file_handle_id:
        client.logger.warning(
            "No validation file handle ID found for this RecordSet. Cannot retrieve detailed validation results."
        )
        return None

    cached_file_path = client.cache.get(
        file_handle_id=self.validation_file_handle_id, path=download_location
    )

    # location in .synapseCache where the file would be corresponding to its FileHandleId
    synapse_cache_location = client.cache.get_cache_dir(
        file_handle_id=self.validation_file_handle_id
    )

    if download_location is not None:
        # Make sure the specified download location is a fully resolved directory
        download_location = ensure_download_location_is_directory(download_location)
    elif cached_file_path is not None:
        # file already cached so use that as the download location
        download_location = os.path.dirname(cached_file_path)
    else:
        # file not cached and no user-specified location so default to .synapseCache
        download_location = synapse_cache_location

    # Generate filename for the validation results CSV
    filename = f"SYNAPSE_RECORDSET_VALIDATION_{self.validation_file_handle_id}.csv"
    destination_path = os.path.join(download_location, filename)

    validation_file_path = await download_by_file_handle(
        file_handle_id=self.validation_file_handle_id,
        synapse_id=self.id,
        entity_type="FileEntity",
        destination=destination_path,
        synapse_client=client,
    )

    return pd.read_csv(validation_file_path)

get_acl_async async

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

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

Note: If the entity does not have local sharing settings, or ACL set directly on it, this will look up the ACL on the benefactor of the entity. The benefactor is the entity that the current entity inherits its permissions from. The benefactor is usually the parent entity, but it can be any ancestor in the hierarchy. For example, a newly created Project will be its own benefactor, while a new FileEntity's benefactor will start off as its containing Project or Folder. If the entity already has local sharing settings, the benefactor would be itself.

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

RETURNS	DESCRIPTION
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,
    check_benefactor: bool = True,
    *,
    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.

    Note: If the entity does not have local sharing settings, or ACL set directly
    on it, this will look up the ACL on the benefactor of the entity. The
    benefactor is the entity that the current entity inherits its permissions from.
    The benefactor is usually the parent entity, but it can be any ancestor in the
    hierarchy. For example, a newly created Project will be its own benefactor,
    while a new FileEntity's benefactor will start off as its containing Project or
    Folder. If the entity already has local sharing settings, the benefactor would
    be itself.

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

    Returns:
        An array containing some combination of
            ['READ', 'UPDATE', 'CREATE', 'DELETE', 'DOWNLOAD', 'MODERATE',
            'CHANGE_PERMISSIONS', 'CHANGE_SETTINGS']
            or an empty array
    """
    return await get_entity_acl_list(
        entity_id=self.id,
        principal_id=str(principal_id) if principal_id is not None else None,
        check_benefactor=check_benefactor,
        synapse_client=synapse_client,
    )

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

import asyncio
from synapseclient import Synapse
from synapseclient.models import File

syn = Synapse()
syn.login()

async def main():
    permissions = await File(id="syn123").get_permissions_async()

asyncio.run(main())

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

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import File

        syn = Synapse()
        syn.login()

        async def main():
            permissions = await File(id="syn123").get_permissions_async()

        asyncio.run(main())
        ```

        Getting access types list from the Permissions object

        ```
        permissions.access_types
        ```
    """
    from synapseclient.core.models.permission import Permissions

    permissions_dict = await get_entity_permissions(
        entity_id=self.id,
        synapse_client=synapse_client,
    )
    return Permissions.from_dict(data=permissions_dict)

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. The term 'benefactor' is used to indicate which Entity an Entity inherits its ACL from. For example, a newly created Project will be its own benefactor, while a new FileEntity's benefactor will start off as its containing Project. If the entity already has local sharing settings the benefactor would be itself. It may also be the immediate parent, somewhere in the parent tree, or the project itself. TYPE: bool DEFAULT: False
warn_if_inherits	When modify_benefactor is True, this does not have any effect. When modify_benefactor is False, and warn_if_inherits is True, a warning log message is produced if the benefactor for the entity you passed into the function is not itself, i.e., it's the parent folder, or another entity in the parent tree. 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 matching https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html.

Setting permissions

Grant all registered users download access

import asyncio
from synapseclient import Synapse
from synapseclient.models import File

syn = Synapse()
syn.login()

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

asyncio.run(main())

Grant the public view access

import asyncio
from synapseclient import Synapse
from synapseclient.models import File

syn = Synapse()
syn.login()

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

asyncio.run(main())

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. The term
            'benefactor' is used to indicate which Entity an Entity inherits its
            ACL from. For example, a newly created Project will be its own
            benefactor, while a new FileEntity's benefactor will start off as its
            containing Project. If the entity already has local sharing settings
            the benefactor would be itself. It may also be the immediate parent,
            somewhere in the parent tree, or the project itself.
        warn_if_inherits: When `modify_benefactor` is True, this does not have any
            effect. When `modify_benefactor` is False, and `warn_if_inherits` is
            True, a warning log message is produced if the benefactor for the
            entity you passed into the function is not itself, i.e., it's the
            parent folder, or another entity in the parent tree.
        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 matching <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/AccessControlList.html>.

    Example: Setting permissions
        Grant all registered users download access

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import File

        syn = Synapse()
        syn.login()

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

        asyncio.run(main())
        ```

        Grant the public view access

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import File

        syn = Synapse()
        syn.login()

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

        asyncio.run(main())
        ```
    """
    if access_type is None:
        access_type = ["READ", "DOWNLOAD"]

    return await set_entity_permissions(
        entity_id=self.id,
        principal_id=str(principal_id) if principal_id is not None else None,
        access_type=access_type,
        modify_benefactor=modify_benefactor,
        warn_if_inherits=warn_if_inherits,
        overwrite=overwrite,
        synapse_client=synapse_client,
    )

delete_permissions_async async

delete_permissions_async(include_self: bool = True, include_container_content: bool = False, recursive: bool = False, target_entity_types: Optional[List[str]] = None, dry_run: bool = False, show_acl_details: bool = True, show_files_in_containers: bool = True, *, synapse_client: Optional[Synapse] = None, _benefactor_tracker: Optional[BenefactorTracker] = None) -> None

Delete the entire Access Control List (ACL) for a given Entity. This is not scoped to a specific user or group, but rather removes all permissions associated with the Entity. After this operation, the Entity will inherit permissions from its benefactor, which is typically its parent entity or the Project it belongs to.

In order to remove permissions for a specific user or group, you should use the set_permissions_async method with the access_type set to an empty list.

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

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

Special notice for Projects: The ACL for a Project cannot be deleted, you must individually update or revoke the permissions for each user or group.

PARAMETER	DESCRIPTION
include_self	If True (default), delete the ACL of the current entity. If False, skip deleting the ACL of the current entity. TYPE: bool DEFAULT: True
include_container_content	If True, delete ACLs from contents directly within containers (files and folders inside self). This must be set to True for recursive to have any effect. Defaults to False. TYPE: bool DEFAULT: False
recursive	If True and the entity is a container (e.g., Project or Folder), recursively process child containers. Note that this must be used with include_container_content=True to have any effect. Setting recursive=True with include_container_content=False will raise a ValueError. Only works on classes that support the sync_from_synapse_async method. TYPE: bool DEFAULT: False
target_entity_types	Specify which entity types to process when deleting ACLs. Allowed values are "folder", "file", "project", "table", "entityview", "materializedview", "virtualtable", "dataset", "datasetcollection", "submissionview" (case-insensitive). If None, defaults to ["folder", "file"]. This does not affect the entity type of the current entity, which is always processed if include_self=True. TYPE: Optional[List[str]] DEFAULT: None
dry_run	If True, log the changes that would be made instead of actually performing the deletions. When enabled, all ACL deletion operations are simulated and logged at info level. Defaults to False. TYPE: bool DEFAULT: False
show_acl_details	When dry_run=True, controls whether current ACL details are displayed for entities that will have their permissions changed. If True (default), shows detailed ACL information. If False, hides ACL details for cleaner output. Has no effect when dry_run=False. TYPE: bool DEFAULT: True
show_files_in_containers	When dry_run=True, controls whether files within containers are displayed in the preview. If True (default), shows all files. If False, hides files when their only change is benefactor inheritance (but still shows files with local ACLs being deleted). Has no effect when dry_run=False. 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
_benefactor_tracker	Internal use tracker for managing benefactor relationships. Used for recursive functionality to track which entities will be affected TYPE: Optional[BenefactorTracker] DEFAULT: None

RETURNS	DESCRIPTION
None	None

RAISES	DESCRIPTION
ValueError	If the entity does not have an ID or if an invalid entity type is provided.
SynapseHTTPError	If there are permission issues or if the entity already inherits permissions.
Exception	For any other errors that may occur during the process.

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

Delete permissions for a single entity

import asyncio
from synapseclient import Synapse
from synapseclient.models import File

syn = Synapse()
syn.login()

async def main():
    await File(id="syn123").delete_permissions_async()

asyncio.run(main())

Delete permissions recursively for a folder and all its children

import asyncio
from synapseclient import Synapse
from synapseclient.models import Folder

syn = Synapse()
syn.login()

async def main():
    # Delete permissions for this folder only (does not affect children)
    await Folder(id="syn123").delete_permissions_async()

    # Delete permissions for all files and folders directly within this folder,
    # but not the folder itself
    await Folder(id="syn123").delete_permissions_async(
        include_self=False,
        include_container_content=True
    )

    # Delete permissions for all items in the entire hierarchy (folders and their files)
    # Both recursive and include_container_content must be True
    await Folder(id="syn123").delete_permissions_async(
        recursive=True,
        include_container_content=True
    )

    # Delete permissions only for folder entities within this folder recursively
    # and their contents
    await Folder(id="syn123").delete_permissions_async(
        recursive=True,
        include_container_content=True,
        target_entity_types=["folder"]
    )

    # Delete permissions only for files within this folder and all subfolders
    await Folder(id="syn123").delete_permissions_async(
        include_self=False,
        recursive=True,
        include_container_content=True,
        target_entity_types=["file"]
    )

    # Delete permissions for specific entity types (e.g., tables and views)
    await Folder(id="syn123").delete_permissions_async(
        recursive=True,
        include_container_content=True,
        target_entity_types=["table", "entityview", "materializedview"]
    )

    # Dry run example: Log what would be deleted without making changes
    await Folder(id="syn123").delete_permissions_async(
        recursive=True,
        include_container_content=True,
        dry_run=True
    )
asyncio.run(main())

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

async def delete_permissions_async(
    self,
    include_self: bool = True,
    include_container_content: bool = False,
    recursive: bool = False,
    target_entity_types: Optional[List[str]] = None,
    dry_run: bool = False,
    show_acl_details: bool = True,
    show_files_in_containers: bool = True,
    *,
    synapse_client: Optional[Synapse] = None,
    _benefactor_tracker: Optional[BenefactorTracker] = None,
) -> None:
    """
    Delete the entire Access Control List (ACL) for a given Entity. This is not
    scoped to a specific user or group, but rather removes all permissions
    associated with the Entity. After this operation, the Entity will inherit
    permissions from its benefactor, which is typically its parent entity or
    the Project it belongs to.

    In order to remove permissions for a specific user or group, you
    should use the `set_permissions_async` method with the `access_type` set to
    an empty list.

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

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

    **Special notice for Projects:** The ACL for a Project cannot be deleted, you
    must individually update or revoke the permissions for each user or group.

    Arguments:
        include_self: If True (default), delete the ACL of the current entity.
            If False, skip deleting the ACL of the current entity.
        include_container_content: If True, delete ACLs from contents directly within
            containers (files and folders inside self). This must be set to
            True for recursive to have any effect. Defaults to False.
        recursive: If True and the entity is a container (e.g., Project or Folder),
            recursively process child containers. Note that this must be used with
            include_container_content=True to have any effect. Setting recursive=True
            with include_container_content=False will raise a ValueError.
            Only works on classes that support the `sync_from_synapse_async` method.
        target_entity_types: Specify which entity types to process when deleting ACLs.
            Allowed values are "folder", "file", "project", "table", "entityview",
            "materializedview", "virtualtable", "dataset", "datasetcollection",
            "submissionview" (case-insensitive). If None, defaults to ["folder", "file"].
            This does not affect the entity type of the current entity, which is always
            processed if `include_self=True`.
        dry_run: If True, log the changes that would be made instead of actually
            performing the deletions. When enabled, all ACL deletion operations are
            simulated and logged at info level. Defaults to False.
        show_acl_details: When dry_run=True, controls whether current ACL details are
            displayed for entities that will have their permissions changed. If True (default),
            shows detailed ACL information. If False, hides ACL details for cleaner output.
            Has no effect when dry_run=False.
        show_files_in_containers: When dry_run=True, controls whether files within containers
            are displayed in the preview. If True (default), shows all files. If False, hides
            files when their only change is benefactor inheritance (but still shows files with
            local ACLs being deleted). Has no effect when dry_run=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.
        _benefactor_tracker: Internal use tracker for managing benefactor relationships.
            Used for recursive functionality to track which entities will be affected

    Returns:
        None

    Raises:
        ValueError: If the entity does not have an ID or if an invalid entity type is provided.
        SynapseHTTPError: If there are permission issues or if the entity already inherits permissions.
        Exception: For any other errors that may occur during the process.

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

    Example: Delete permissions for a single entity
        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import File

        syn = Synapse()
        syn.login()

        async def main():
            await File(id="syn123").delete_permissions_async()

        asyncio.run(main())
        ```

    Example: Delete permissions recursively for a folder and all its children
        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import Folder

        syn = Synapse()
        syn.login()

        async def main():
            # Delete permissions for this folder only (does not affect children)
            await Folder(id="syn123").delete_permissions_async()

            # Delete permissions for all files and folders directly within this folder,
            # but not the folder itself
            await Folder(id="syn123").delete_permissions_async(
                include_self=False,
                include_container_content=True
            )

            # Delete permissions for all items in the entire hierarchy (folders and their files)
            # Both recursive and include_container_content must be True
            await Folder(id="syn123").delete_permissions_async(
                recursive=True,
                include_container_content=True
            )

            # Delete permissions only for folder entities within this folder recursively
            # and their contents
            await Folder(id="syn123").delete_permissions_async(
                recursive=True,
                include_container_content=True,
                target_entity_types=["folder"]
            )

            # Delete permissions only for files within this folder and all subfolders
            await Folder(id="syn123").delete_permissions_async(
                include_self=False,
                recursive=True,
                include_container_content=True,
                target_entity_types=["file"]
            )

            # Delete permissions for specific entity types (e.g., tables and views)
            await Folder(id="syn123").delete_permissions_async(
                recursive=True,
                include_container_content=True,
                target_entity_types=["table", "entityview", "materializedview"]
            )

            # Dry run example: Log what would be deleted without making changes
            await Folder(id="syn123").delete_permissions_async(
                recursive=True,
                include_container_content=True,
                dry_run=True
            )
        asyncio.run(main())
        ```
    """
    if not self.id:
        raise ValueError("The entity must have an ID to delete permissions.")

    client = Synapse.get_client(synapse_client=synapse_client)

    if include_self and self.__class__.__name__.lower() == "project":
        client.logger.warning(
            "The ACL for a Project cannot be deleted, you must individually update or "
            "revoke the permissions for each user or group. Continuing without deleting "
            "the Project's ACL."
        )
        include_self = False

    normalized_types = self._normalize_target_entity_types(target_entity_types)

    is_top_level = not _benefactor_tracker
    benefactor_tracker = _benefactor_tracker or BenefactorTracker()

    should_process_children = (recursive or include_container_content) and hasattr(
        self, "sync_from_synapse_async"
    )
    all_entities = [self] if include_self else []

    custom_message = "Deleting ACLs [Dry Run]..." if dry_run else "Deleting ACLs..."
    with shared_download_progress_bar(
        file_size=1, synapse_client=client, custom_message=custom_message, unit=None
    ) as progress_bar:
        if progress_bar:
            progress_bar.update(1)  # Initial setup complete

        if should_process_children:
            if recursive and not include_container_content:
                raise ValueError(
                    "When recursive=True, include_container_content must also be True. "
                    "Setting recursive=True with include_container_content=False has no effect."
                )

            if progress_bar:
                progress_bar.total += 1
                progress_bar.refresh()

            all_entities = await self._collect_entities(
                client=client,
                target_entity_types=normalized_types,
                include_container_content=include_container_content,
                recursive=recursive,
                progress_bar=progress_bar,
            )
            if progress_bar:
                progress_bar.update(1)

            entity_ids = [entity.id for entity in all_entities if entity.id]
            if entity_ids:
                if progress_bar:
                    progress_bar.total += 1
                    progress_bar.refresh()
                await benefactor_tracker.track_entity_benefactor(
                    entity_ids=entity_ids,
                    synapse_client=client,
                    progress_bar=progress_bar,
                )
            else:
                if progress_bar:
                    progress_bar.total += 1
                    progress_bar.refresh()
                    progress_bar.update(1)

        if is_top_level:
            if progress_bar:
                progress_bar.total += 1
                progress_bar.refresh()
            await self._build_and_log_run_tree(
                client=client,
                benefactor_tracker=benefactor_tracker,
                collected_entities=all_entities,
                include_self=include_self,
                show_acl_details=show_acl_details,
                show_files_in_containers=show_files_in_containers,
                progress_bar=progress_bar,
                dry_run=dry_run,
            )

        if dry_run:
            return

        if include_self:
            if progress_bar:
                progress_bar.total += 1
                progress_bar.refresh()
            await self._delete_current_entity_acl(
                client=client,
                benefactor_tracker=benefactor_tracker,
                progress_bar=progress_bar,
            )

        if should_process_children:
            if include_container_content:
                if progress_bar:
                    progress_bar.total += 1
                    progress_bar.refresh()
                await self._process_container_contents(
                    client=client,
                    target_entity_types=normalized_types,
                    benefactor_tracker=benefactor_tracker,
                    progress_bar=progress_bar,
                    recursive=recursive,
                    include_container_content=include_container_content,
                )
                if progress_bar:
                    progress_bar.update(1)  # Process container contents complete

list_acl_async async

list_acl_async(recursive: bool = False, include_container_content: bool = False, target_entity_types: Optional[List[str]] = None, log_tree: bool = False, *, synapse_client: Optional[Synapse] = None, _progress_bar: Optional[tqdm] = None) -> AclListResult

List the Access Control Lists (ACLs) for this entity and optionally its children.

This function returns the local sharing settings for the entity and optionally its children. It provides a mapping of all ACLs for the given container/entity.

Important Note: This function returns the LOCAL sharing settings only, not the effective permissions that each Synapse User ID/Team has on the entities. More permissive permissions could be granted via a Team that the user has access to that has permissions on the entity, or through inheritance from parent entities.

PARAMETER	DESCRIPTION
recursive	If True and the entity is a container (e.g., Project or Folder), recursively process child containers. Note that this must be used with include_container_content=True to have any effect. Setting recursive=True with include_container_content=False will raise a ValueError. Only works on classes that support the sync_from_synapse_async method. TYPE: bool DEFAULT: False
include_container_content	If True, include ACLs from contents directly within containers (files and folders inside self). This must be set to True for recursive to have any effect. Defaults to False. TYPE: bool DEFAULT: False
target_entity_types	Specify which entity types to process when listing ACLs. Allowed values are "folder", "file", "project", "table", "entityview", "materializedview", "virtualtable", "dataset", "datasetcollection", "submissionview" (case-insensitive). If None, defaults to ["folder", "file"]. TYPE: Optional[List[str]] DEFAULT: None
log_tree	If True, logs the ACL results to console in ASCII tree format showing entity hierarchies and their ACL permissions in a tree-like structure. Defaults to False. TYPE: bool DEFAULT: False
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None
_progress_bar	Internal parameter. Progress bar instance to use for updates when called recursively. Should not be used by external callers. TYPE: Optional[tqdm] DEFAULT: None

RETURNS	DESCRIPTION
AclListResult	An AclListResult object containing a structured representation of ACLs where:
AclListResult	entity_acls: A list of EntityAcl objects, each representing one entity's ACL
AclListResult	Each EntityAcl contains acl_entries (a list of AclEntry objects)
AclListResult	Each AclEntry contains the principal_id and their list of permissions

RAISES	DESCRIPTION
ValueError	If the entity does not have an ID or if an invalid entity type is provided.
SynapseHTTPError	If there are permission issues accessing ACLs.
Exception	For any other errors that may occur during the process.

List ACLs for a single entity

import asyncio
from synapseclient import Synapse
from synapseclient.models import File

syn = Synapse()
syn.login()

async def main():
    acl_result = await File(id="syn123").list_acl_async()
    print(acl_result)

    # Access entity ACLs (entity_acls is a list, not a dict)
    for entity_acl in acl_result.all_entity_acls:
        if entity_acl.entity_id == "syn123":
            # Access individual ACL entries
            for acl_entry in entity_acl.acl_entries:
                if acl_entry.principal_id == "273948":
                    print(f"Principal 273948 has permissions: {acl_entry.permissions}")

    # I can also access the ACL for the file itself
    print(acl_result.entity_acl)

    print(acl_result)

asyncio.run(main())

List ACLs recursively for a folder and all its children

import asyncio
from synapseclient import Synapse
from synapseclient.models import Folder

syn = Synapse()
syn.login()

async def main():
    acl_result = await Folder(id="syn123").list_acl_async(
        recursive=True,
        include_container_content=True
    )

    # Access each entity's ACL (entity_acls is a list)
    for entity_acl in acl_result.all_entity_acls:
        print(f"Entity {entity_acl.entity_id} has ACL with {len(entity_acl.acl_entries)} principals")

    # I can also access the ACL for the folder itself
    print(acl_result.entity_acl)

    # List ACLs for only folder entities
    folder_acl_result = await Folder(id="syn123").list_acl_async(
        recursive=True,
        include_container_content=True,
        target_entity_types=["folder"]
    )

    # List ACLs for specific entity types (e.g., tables and views)
    table_view_acl_result = await Folder(id="syn123").list_acl_async(
        recursive=True,
        include_container_content=True,
        target_entity_types=["table", "entityview", "materializedview"]
    )

asyncio.run(main())

List ACLs with ASCII tree visualization

When log_tree=True, the ACLs will be logged in a tree format. Additionally, the ascii_tree attribute of the AclListResult will contain the ASCII tree representation of the ACLs.

import asyncio
from synapseclient import Synapse
from synapseclient.models import Folder

syn = Synapse()
syn.login()

async def main():
    acl_result = await Folder(id="syn123").list_acl_async(
        recursive=True,
        include_container_content=True,
        log_tree=True, # Enable ASCII tree logging
    )

    # The ASCII tree representation of the ACLs will also be available
    # in acl_result.ascii_tree
    print(acl_result.ascii_tree)

asyncio.run(main())

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

async def list_acl_async(
    self,
    recursive: bool = False,
    include_container_content: bool = False,
    target_entity_types: Optional[List[str]] = None,
    log_tree: bool = False,
    *,
    synapse_client: Optional[Synapse] = None,
    _progress_bar: Optional[tqdm] = None,  # Internal parameter for recursive calls
) -> AclListResult:
    """
    List the Access Control Lists (ACLs) for this entity and optionally its children.

    This function returns the local sharing settings for the entity and optionally
    its children. It provides a mapping of all ACLs for the given container/entity.

    **Important Note:** This function returns the LOCAL sharing settings only, not
    the effective permissions that each Synapse User ID/Team has on the entities.
    More permissive permissions could be granted via a Team that the user has access
    to that has permissions on the entity, or through inheritance from parent entities.

    Arguments:
        recursive: If True and the entity is a container (e.g., Project or Folder),
            recursively process child containers. Note that this must be used with
            include_container_content=True to have any effect. Setting recursive=True
            with include_container_content=False will raise a ValueError.
            Only works on classes that support the `sync_from_synapse_async` method.
        include_container_content: If True, include ACLs from contents directly within
            containers (files and folders inside self). This must be set to
            True for recursive to have any effect. Defaults to False.
        target_entity_types: Specify which entity types to process when listing ACLs.
            Allowed values are "folder", "file", "project", "table", "entityview",
            "materializedview", "virtualtable", "dataset", "datasetcollection",
            "submissionview" (case-insensitive). If None, defaults to ["folder", "file"].
        log_tree: If True, logs the ACL results to console in ASCII tree format showing
            entity hierarchies and their ACL permissions in a tree-like structure.
            Defaults to False.
        synapse_client: If not passed in and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.
        _progress_bar: Internal parameter. Progress bar instance to use for updates
            when called recursively. Should not be used by external callers.

    Returns:
        An AclListResult object containing a structured representation of ACLs where:
        - entity_acls: A list of EntityAcl objects, each representing one entity's ACL
        - Each EntityAcl contains acl_entries (a list of AclEntry objects)
        - Each AclEntry contains the principal_id and their list of permissions

    Raises:
        ValueError: If the entity does not have an ID or if an invalid entity type is provided.
        SynapseHTTPError: If there are permission issues accessing ACLs.
        Exception: For any other errors that may occur during the process.

    Example: List ACLs for a single entity
        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import File

        syn = Synapse()
        syn.login()

        async def main():
            acl_result = await File(id="syn123").list_acl_async()
            print(acl_result)

            # Access entity ACLs (entity_acls is a list, not a dict)
            for entity_acl in acl_result.all_entity_acls:
                if entity_acl.entity_id == "syn123":
                    # Access individual ACL entries
                    for acl_entry in entity_acl.acl_entries:
                        if acl_entry.principal_id == "273948":
                            print(f"Principal 273948 has permissions: {acl_entry.permissions}")

            # I can also access the ACL for the file itself
            print(acl_result.entity_acl)

            print(acl_result)

        asyncio.run(main())
        ```

    Example: List ACLs recursively for a folder and all its children
        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import Folder

        syn = Synapse()
        syn.login()

        async def main():
            acl_result = await Folder(id="syn123").list_acl_async(
                recursive=True,
                include_container_content=True
            )

            # Access each entity's ACL (entity_acls is a list)
            for entity_acl in acl_result.all_entity_acls:
                print(f"Entity {entity_acl.entity_id} has ACL with {len(entity_acl.acl_entries)} principals")

            # I can also access the ACL for the folder itself
            print(acl_result.entity_acl)

            # List ACLs for only folder entities
            folder_acl_result = await Folder(id="syn123").list_acl_async(
                recursive=True,
                include_container_content=True,
                target_entity_types=["folder"]
            )

            # List ACLs for specific entity types (e.g., tables and views)
            table_view_acl_result = await Folder(id="syn123").list_acl_async(
                recursive=True,
                include_container_content=True,
                target_entity_types=["table", "entityview", "materializedview"]
            )

        asyncio.run(main())
        ```

    Example: List ACLs with ASCII tree visualization
        When `log_tree=True`, the ACLs will be logged in a tree format. Additionally,
        the `ascii_tree` attribute of the AclListResult will contain the ASCII tree
        representation of the ACLs.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import Folder

        syn = Synapse()
        syn.login()

        async def main():
            acl_result = await Folder(id="syn123").list_acl_async(
                recursive=True,
                include_container_content=True,
                log_tree=True, # Enable ASCII tree logging
            )

            # The ASCII tree representation of the ACLs will also be available
            # in acl_result.ascii_tree
            print(acl_result.ascii_tree)

        asyncio.run(main())
        ```
    """
    if not self.id:
        raise ValueError("The entity must have an ID to list ACLs.")

    normalized_types = self._normalize_target_entity_types(target_entity_types)
    client = Synapse.get_client(synapse_client=synapse_client)

    all_acls: Dict[str, Dict[str, List[str]]] = {}
    all_entities = []

    # Only update progress bar for self ACL if we're the top-level call (not recursive)
    # When _progress_bar is passed, it means this is a recursive call and the parent
    # is managing progress updates
    update_progress_for_self = _progress_bar is None
    acl = await self._get_current_entity_acl(
        client=client,
        progress_bar=_progress_bar if update_progress_for_self else None,
    )
    if acl is not None:
        all_acls[self.id] = acl
    all_entities.append(self)

    should_process_children = (recursive or include_container_content) and hasattr(
        self, "sync_from_synapse_async"
    )

    if should_process_children and (recursive and not include_container_content):
        raise ValueError(
            "When recursive=True, include_container_content must also be True. "
            "Setting recursive=True with include_container_content=False has no effect."
        )

    if should_process_children and _progress_bar is None:
        with shared_download_progress_bar(
            file_size=1,
            synapse_client=client,
            custom_message="Collecting ACLs...",
            unit=None,
        ) as progress_bar:
            await self._process_children_with_progress(
                client=client,
                normalized_types=normalized_types,
                include_container_content=include_container_content,
                recursive=recursive,
                all_entities=all_entities,
                all_acls=all_acls,
                progress_bar=progress_bar,
            )
            # Ensure progress bar reaches 100% completion
            if progress_bar:
                remaining = (
                    progress_bar.total - progress_bar.n
                    if progress_bar.total > progress_bar.n
                    else 0
                )
                if remaining > 0:
                    progress_bar.update(remaining)
    elif should_process_children:
        await self._process_children_with_progress(
            client=client,
            normalized_types=normalized_types,
            include_container_content=include_container_content,
            recursive=recursive,
            all_entities=all_entities,
            all_acls=all_acls,
            progress_bar=_progress_bar,
        )
    current_acl = all_acls.get(self.id)
    acl_result = AclListResult.from_dict(
        all_acl_dict=all_acls, current_acl_dict=current_acl
    )

    if log_tree:
        logged_tree = await self._log_acl_tree(acl_result, all_entities, client)
        acl_result.ascii_tree = logged_tree

    return acl_result

bind_schema_async async

bind_schema_async(json_schema_uri: str, *, enable_derived_annotations: bool = False, synapse_client: Optional[Synapse] = None) -> JSONSchemaBinding

Bind a JSON schema to the entity.

PARAMETER	DESCRIPTION
json_schema_uri	The URI of the JSON schema to bind to the entity. TYPE: str
enable_derived_annotations	If true, enable derived annotations. Defaults to False. TYPE: bool DEFAULT: False
synapse_client	The Synapse client instance. If not provided, the last created instance from the Synapse class constructor will be used. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
JSONSchemaBinding	An object containing details about the JSON schema binding.

Using this function

Binding JSON schema to a folder or a file. This example expects that you have a Synapse project to use, and a file to upload. Set the PROJECT_NAME and FILE_PATH variables to your project name and file path respectively.

import asyncio
from synapseclient import Synapse
from synapseclient.models import File, Folder

syn = Synapse()
syn.login()

# Define Project and JSON schema info
PROJECT_NAME = "test_json_schema_project"  # replace with your project name
FILE_PATH = "~/Sample.txt"  # replace with your test file path

PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
ORG_NAME = "UniqueOrg"  # replace with your organization name
SCHEMA_NAME = "myTestSchema"  # replace with your schema name
FOLDER_NAME = "test_script_folder"
VERSION = "0.0.1"
SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

# Create organization (if not already created)
js = syn.service("json_schema")
all_orgs = js.list_organizations()
for org in all_orgs:
    if org["name"] == ORG_NAME:
        print(f"Organization {ORG_NAME} already exists: {org}")
        break
else:
    print(f"Creating organization {ORG_NAME}.")
    created_organization = js.create_organization(ORG_NAME)
    print(f"Created organization: {created_organization}")

my_test_org = js.JsonSchemaOrganization(ORG_NAME)
test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

if not test_schema:
    # Create the schema (if not already created)
    schema_definition = {
        "$id": "mySchema",
        "type": "object",
        "properties": {
            "foo": {"type": "string"},
            "bar": {"type": "integer"},
        },
        "required": ["foo"]
    }
    test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
    print(f"Created new schema: {SCHEMA_NAME}")

async def main():
    # Create a test folder
    test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
    await test_folder.store_async()
    print(f"Created test folder: {FOLDER_NAME}")

    # Bind JSON schema to the folder
    bound_schema = await test_folder.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Result from binding schema to folder: {bound_schema}")

    # Create and bind schema to a file
    example_file = File(
        path=FILE_PATH,  # Replace with your test file path
        parent_id=test_folder.id,
    )
    await example_file.store_async()

    bound_schema_file = await example_file.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Result from binding schema to file: {bound_schema_file}")

asyncio.run(main())

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

async def bind_schema_async(
    self,
    json_schema_uri: str,
    *,
    enable_derived_annotations: bool = False,
    synapse_client: Optional["Synapse"] = None,
) -> JSONSchemaBinding:
    """
    Bind a JSON schema to the entity.

    Arguments:
        json_schema_uri: The URI of the JSON schema to bind to the entity.
        enable_derived_annotations: If true, enable derived annotations. Defaults to False.
        synapse_client: The Synapse client instance. If not provided,
            the last created instance from the Synapse class constructor will be used.

    Returns:
        An object containing details about the JSON schema binding.

    Example: Using this function
        Binding JSON schema to a folder or a file. This example expects that you
        have a Synapse project to use, and a file to upload. Set the `PROJECT_NAME`
        and `FILE_PATH` variables to your project name and file path respectively.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import File, Folder

        syn = Synapse()
        syn.login()

        # Define Project and JSON schema info
        PROJECT_NAME = "test_json_schema_project"  # replace with your project name
        FILE_PATH = "~/Sample.txt"  # replace with your test file path

        PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
        ORG_NAME = "UniqueOrg"  # replace with your organization name
        SCHEMA_NAME = "myTestSchema"  # replace with your schema name
        FOLDER_NAME = "test_script_folder"
        VERSION = "0.0.1"
        SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

        # Create organization (if not already created)
        js = syn.service("json_schema")
        all_orgs = js.list_organizations()
        for org in all_orgs:
            if org["name"] == ORG_NAME:
                print(f"Organization {ORG_NAME} already exists: {org}")
                break
        else:
            print(f"Creating organization {ORG_NAME}.")
            created_organization = js.create_organization(ORG_NAME)
            print(f"Created organization: {created_organization}")

        my_test_org = js.JsonSchemaOrganization(ORG_NAME)
        test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

        if not test_schema:
            # Create the schema (if not already created)
            schema_definition = {
                "$id": "mySchema",
                "type": "object",
                "properties": {
                    "foo": {"type": "string"},
                    "bar": {"type": "integer"},
                },
                "required": ["foo"]
            }
            test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
            print(f"Created new schema: {SCHEMA_NAME}")

        async def main():
            # Create a test folder
            test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
            await test_folder.store_async()
            print(f"Created test folder: {FOLDER_NAME}")

            # Bind JSON schema to the folder
            bound_schema = await test_folder.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Result from binding schema to folder: {bound_schema}")

            # Create and bind schema to a file
            example_file = File(
                path=FILE_PATH,  # Replace with your test file path
                parent_id=test_folder.id,
            )
            await example_file.store_async()

            bound_schema_file = await example_file.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Result from binding schema to file: {bound_schema_file}")

        asyncio.run(main())
        ```
    """
    response = await bind_json_schema_to_entity(
        synapse_id=self.id,
        json_schema_uri=json_schema_uri,
        enable_derived_annotations=enable_derived_annotations,
        synapse_client=synapse_client,
    )
    json_schema_version = response.get("jsonSchemaVersionInfo", {})
    return JSONSchemaBinding(
        json_schema_version_info=JSONSchemaVersionInfo(
            organization_id=json_schema_version.get("organizationId", None),
            organization_name=json_schema_version.get("organizationName", None),
            schema_id=json_schema_version.get("schemaId", None),
            id=json_schema_version.get("$id", None),
            schema_name=json_schema_version.get("schemaName", None),
            version_id=json_schema_version.get("versionId", None),
            semantic_version=json_schema_version.get("semanticVersion", None),
            json_sha256_hex=json_schema_version.get("jsonSHA256Hex", None),
            created_on=json_schema_version.get("createdOn", None),
            created_by=json_schema_version.get("createdBy", None),
        ),
        object_id=response.get("objectId", None),
        object_type=response.get("objectType", None),
        created_on=response.get("createdOn", None),
        created_by=response.get("createdBy", None),
        enable_derived_annotations=response.get("enableDerivedAnnotations", None),
    )

get_schema_async async

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

Get the JSON schema bound to the entity.

PARAMETER	DESCRIPTION
synapse_client	The Synapse client instance. If not provided, the last created instance from the Synapse class constructor will be used. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
JSONSchemaBinding	An object containing details about the bound JSON schema.

Using this function

Retrieving the bound JSON schema from a folder or file. This example demonstrates how to get existing schema bindings from entities that already have schemas bound. Set the PROJECT_NAME and FILE_PATH variables to your project name and file path respectively.

import asyncio
from synapseclient import Synapse
from synapseclient.models import File, Folder

syn = Synapse()
syn.login()

# Define Project and JSON schema info
PROJECT_NAME = "test_json_schema_project"  # replace with your project name
FILE_PATH = "~/Sample.txt"  # replace with your test file path

PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
ORG_NAME = "UniqueOrg"  # replace with your organization name
SCHEMA_NAME = "myTestSchema"  # replace with your schema name
FOLDER_NAME = "test_script_folder"
VERSION = "0.0.1"
SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

# Create organization (if not already created)
js = syn.service("json_schema")
all_orgs = js.list_organizations()
for org in all_orgs:
    if org["name"] == ORG_NAME:
        print(f"Organization {ORG_NAME} already exists: {org}")
        break
else:
    print(f"Creating organization {ORG_NAME}.")
    created_organization = js.create_organization(ORG_NAME)
    print(f"Created organization: {created_organization}")

my_test_org = js.JsonSchemaOrganization(ORG_NAME)
test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

if not test_schema:
    # Create the schema (if not already created)
    schema_definition = {
        "$id": "mySchema",
        "type": "object",
        "properties": {
            "foo": {"type": "string"},
            "bar": {"type": "integer"},
        },
        "required": ["foo"]
    }
    test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
    print(f"Created new schema: {SCHEMA_NAME}")

async def main():
    # Create a test folder
    test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
    await test_folder.store_async()
    print(f"Created test folder: {FOLDER_NAME}")

    # Bind JSON schema to the folder first
    bound_schema = await test_folder.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Bound schema to folder: {bound_schema}")

    # Create and bind schema to a file
    example_file = File(
        path=FILE_PATH,  # Replace with your test file path
        parent_id=test_folder.id,
    )
    await example_file.store_async()

    bound_schema_file = await example_file.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Bound schema to file: {bound_schema_file}")

    # Retrieve the bound schema from the folder
    bound_schema = await test_folder.get_schema_async()
    print(f"Retrieved schema from folder: {bound_schema}")

    # Retrieve the bound schema from the file
    bound_schema_file = await example_file.get_schema_async()
    print(f"Retrieved schema from file: {bound_schema_file}")

asyncio.run(main())

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

async def get_schema_async(
    self, *, synapse_client: Optional["Synapse"] = None
) -> JSONSchemaBinding:
    """
    Get the JSON schema bound to the entity.

    Arguments:
        synapse_client: The Synapse client instance. If not provided,
            the last created instance from the Synapse class constructor will be used.

    Returns:
        An object containing details about the bound JSON schema.

    Example: Using this function
        Retrieving the bound JSON schema from a folder or file. This example demonstrates
        how to get existing schema bindings from entities that already have schemas bound.
        Set the `PROJECT_NAME` and `FILE_PATH` variables to your project name
        and file path respectively.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import File, Folder

        syn = Synapse()
        syn.login()

        # Define Project and JSON schema info
        PROJECT_NAME = "test_json_schema_project"  # replace with your project name
        FILE_PATH = "~/Sample.txt"  # replace with your test file path

        PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
        ORG_NAME = "UniqueOrg"  # replace with your organization name
        SCHEMA_NAME = "myTestSchema"  # replace with your schema name
        FOLDER_NAME = "test_script_folder"
        VERSION = "0.0.1"
        SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

        # Create organization (if not already created)
        js = syn.service("json_schema")
        all_orgs = js.list_organizations()
        for org in all_orgs:
            if org["name"] == ORG_NAME:
                print(f"Organization {ORG_NAME} already exists: {org}")
                break
        else:
            print(f"Creating organization {ORG_NAME}.")
            created_organization = js.create_organization(ORG_NAME)
            print(f"Created organization: {created_organization}")

        my_test_org = js.JsonSchemaOrganization(ORG_NAME)
        test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

        if not test_schema:
            # Create the schema (if not already created)
            schema_definition = {
                "$id": "mySchema",
                "type": "object",
                "properties": {
                    "foo": {"type": "string"},
                    "bar": {"type": "integer"},
                },
                "required": ["foo"]
            }
            test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
            print(f"Created new schema: {SCHEMA_NAME}")

        async def main():
            # Create a test folder
            test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
            await test_folder.store_async()
            print(f"Created test folder: {FOLDER_NAME}")

            # Bind JSON schema to the folder first
            bound_schema = await test_folder.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Bound schema to folder: {bound_schema}")

            # Create and bind schema to a file
            example_file = File(
                path=FILE_PATH,  # Replace with your test file path
                parent_id=test_folder.id,
            )
            await example_file.store_async()

            bound_schema_file = await example_file.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Bound schema to file: {bound_schema_file}")

            # Retrieve the bound schema from the folder
            bound_schema = await test_folder.get_schema_async()
            print(f"Retrieved schema from folder: {bound_schema}")

            # Retrieve the bound schema from the file
            bound_schema_file = await example_file.get_schema_async()
            print(f"Retrieved schema from file: {bound_schema_file}")

        asyncio.run(main())
        ```
    """
    response = await get_json_schema_from_entity(
        synapse_id=self.id, synapse_client=synapse_client
    )
    json_schema_version_info = response.get("jsonSchemaVersionInfo", {})
    return JSONSchemaBinding(
        json_schema_version_info=JSONSchemaVersionInfo(
            organization_id=json_schema_version_info.get("organizationId", None),
            organization_name=json_schema_version_info.get(
                "organizationName", None
            ),
            schema_id=json_schema_version_info.get("schemaId", None),
            id=json_schema_version_info.get("$id", None),
            schema_name=json_schema_version_info.get("schemaName", None),
            version_id=json_schema_version_info.get("versionId", None),
            semantic_version=json_schema_version_info.get("semanticVersion", None),
            json_sha256_hex=json_schema_version_info.get("jsonSHA256Hex", None),
            created_on=json_schema_version_info.get("createdOn", None),
            created_by=json_schema_version_info.get("createdBy", None),
        ),
        object_id=response.get("objectId", None),
        object_type=response.get("objectType", None),
        created_on=response.get("createdOn", None),
        created_by=response.get("createdBy", None),
        enable_derived_annotations=response.get("enableDerivedAnnotations", None),
    )

unbind_schema_async async

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

Unbind the JSON schema bound to the entity.

PARAMETER	DESCRIPTION
synapse_client	The Synapse client instance. If not provided, the last created instance from the Synapse class constructor will be used. TYPE: Optional[Synapse] DEFAULT: None

Using this function

Unbinding a JSON schema from a folder or file. This example demonstrates how to remove schema bindings from entities. Assumes entities already have schemas bound. Set the PROJECT_NAME and FILE_PATH variables to your project name and file path respectively.

import asyncio
from synapseclient import Synapse
from synapseclient.models import File, Folder

syn = Synapse()
syn.login()

# Define Project and JSON schema info
PROJECT_NAME = "test_json_schema_project"  # replace with your project name
FILE_PATH = "~/Sample.txt"  # replace with your test file path

PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
ORG_NAME = "UniqueOrg"  # replace with your organization name
SCHEMA_NAME = "myTestSchema"  # replace with your schema name
FOLDER_NAME = "test_script_folder"
VERSION = "0.0.1"
SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

# Create organization (if not already created)
js = syn.service("json_schema")
all_orgs = js.list_organizations()
for org in all_orgs:
    if org["name"] == ORG_NAME:
        print(f"Organization {ORG_NAME} already exists: {org}")
        break
else:
    print(f"Creating organization {ORG_NAME}.")
    created_organization = js.create_organization(ORG_NAME)
    print(f"Created organization: {created_organization}")

my_test_org = js.JsonSchemaOrganization(ORG_NAME)
test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

if not test_schema:
    # Create the schema (if not already created)
    schema_definition = {
        "$id": "mySchema",
        "type": "object",
        "properties": {
            "foo": {"type": "string"},
            "bar": {"type": "integer"},
        },
        "required": ["foo"]
    }
    test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
    print(f"Created new schema: {SCHEMA_NAME}")

async def main():
    # Create a test folder
    test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
    await test_folder.store_async()
    print(f"Created test folder: {FOLDER_NAME}")

    # Bind JSON schema to the folder first
    bound_schema = await test_folder.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Bound schema to folder: {bound_schema}")

    # Create and bind schema to a file
    example_file = File(
        path=FILE_PATH,  # Replace with your test file path
        parent_id=test_folder.id,
    )
    await example_file.store_async()
    print(f"Created test file: {FILE_PATH}")

    bound_schema_file = await example_file.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Bound schema to file: {bound_schema_file}")

    # Unbind the schema from the folder
    await test_folder.unbind_schema_async()
    print("Successfully unbound schema from folder")

    # Unbind the schema from the file
    await example_file.unbind_schema_async()
    print("Successfully unbound schema from file")

asyncio.run(main())

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

async def unbind_schema_async(
    self, *, synapse_client: Optional["Synapse"] = None
) -> None:
    """
    Unbind the JSON schema bound to the entity.

    Arguments:
        synapse_client: The Synapse client instance. If not provided,
            the last created instance from the Synapse class constructor will be used.

    Example: Using this function
        Unbinding a JSON schema from a folder or file. This example demonstrates
        how to remove schema bindings from entities. Assumes entities already have
        schemas bound. Set the `PROJECT_NAME` and `FILE_PATH` variables to your
        project name and file path respectively.


        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import File, Folder

        syn = Synapse()
        syn.login()

        # Define Project and JSON schema info
        PROJECT_NAME = "test_json_schema_project"  # replace with your project name
        FILE_PATH = "~/Sample.txt"  # replace with your test file path

        PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
        ORG_NAME = "UniqueOrg"  # replace with your organization name
        SCHEMA_NAME = "myTestSchema"  # replace with your schema name
        FOLDER_NAME = "test_script_folder"
        VERSION = "0.0.1"
        SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

        # Create organization (if not already created)
        js = syn.service("json_schema")
        all_orgs = js.list_organizations()
        for org in all_orgs:
            if org["name"] == ORG_NAME:
                print(f"Organization {ORG_NAME} already exists: {org}")
                break
        else:
            print(f"Creating organization {ORG_NAME}.")
            created_organization = js.create_organization(ORG_NAME)
            print(f"Created organization: {created_organization}")

        my_test_org = js.JsonSchemaOrganization(ORG_NAME)
        test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

        if not test_schema:
            # Create the schema (if not already created)
            schema_definition = {
                "$id": "mySchema",
                "type": "object",
                "properties": {
                    "foo": {"type": "string"},
                    "bar": {"type": "integer"},
                },
                "required": ["foo"]
            }
            test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
            print(f"Created new schema: {SCHEMA_NAME}")

        async def main():
            # Create a test folder
            test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
            await test_folder.store_async()
            print(f"Created test folder: {FOLDER_NAME}")

            # Bind JSON schema to the folder first
            bound_schema = await test_folder.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Bound schema to folder: {bound_schema}")

            # Create and bind schema to a file
            example_file = File(
                path=FILE_PATH,  # Replace with your test file path
                parent_id=test_folder.id,
            )
            await example_file.store_async()
            print(f"Created test file: {FILE_PATH}")

            bound_schema_file = await example_file.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Bound schema to file: {bound_schema_file}")

            # Unbind the schema from the folder
            await test_folder.unbind_schema_async()
            print("Successfully unbound schema from folder")

            # Unbind the schema from the file
            await example_file.unbind_schema_async()
            print("Successfully unbound schema from file")

        asyncio.run(main())
        ```
    """
    return await delete_json_schema_from_entity(
        synapse_id=self.id, synapse_client=synapse_client
    )

validate_schema_async async

validate_schema_async(*, synapse_client: Optional[Synapse] = None) -> Union[JSONSchemaValidation, InvalidJSONSchemaValidation]

Validate the entity against the bound JSON schema.

PARAMETER	DESCRIPTION
synapse_client	The Synapse client instance. If not provided, the last created instance from the Synapse class constructor will be used. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
Union[JSONSchemaValidation, InvalidJSONSchemaValidation]	The validation results.

Using this function

Validating a folder or file against the bound JSON schema. This example demonstrates how to validate entities with annotations against their bound schemas. Requires entities to have schemas already bound. Set the PROJECT_NAME and FILE_PATH variables to your project name and file path respectively.

import asyncio
import time
from synapseclient import Synapse
from synapseclient.models import File, Folder

syn = Synapse()
syn.login()

# Define Project and JSON schema info
PROJECT_NAME = "test_json_schema_project"  # replace with your project name
FILE_PATH = "~/Sample.txt"  # replace with your test file path

PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
ORG_NAME = "UniqueOrg"  # replace with your organization name
SCHEMA_NAME = "myTestSchema"  # replace with your schema name
FOLDER_NAME = "test_script_folder"
VERSION = "0.0.1"
SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

# Create organization (if not already created)
js = syn.service("json_schema")
all_orgs = js.list_organizations()
for org in all_orgs:
    if org["name"] == ORG_NAME:
        print(f"Organization {ORG_NAME} already exists: {org}")
        break
else:
    print(f"Creating organization {ORG_NAME}.")
    created_organization = js.create_organization(ORG_NAME)
    print(f"Created organization: {created_organization}")

my_test_org = js.JsonSchemaOrganization(ORG_NAME)
test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

if not test_schema:
    # Create the schema (if not already created)
    schema_definition = {
        "$id": "mySchema",
        "type": "object",
        "properties": {
            "foo": {"type": "string"},
            "bar": {"type": "integer"},
        },
        "required": ["foo"]
    }
    test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
    print(f"Created new schema: {SCHEMA_NAME}")

async def main():
    # Create a test folder
    test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
    await test_folder.store_async()
    print(f"Created test folder: {FOLDER_NAME}")

    # Bind JSON schema to the folder
    bound_schema = await test_folder.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Bound schema to folder: {bound_schema}")

    # Create and bind schema to a file
    example_file = File(
        path=FILE_PATH,  # Replace with your test file path
        parent_id=test_folder.id,
    )
    await example_file.store_async()

    bound_schema_file = await example_file.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Bound schema to file: {bound_schema_file}")

    # Validate the folder entity against the bound schema
    test_folder.annotations = {"foo": "test_value", "bar": 42}  # Example annotations
    await test_folder.store_async()
    print("Added annotations to folder and stored")
    time.sleep(2)  # Allow time for processing

    validation_response = await test_folder.validate_schema_async()
    print(f"Folder validation response: {validation_response}")

    # Validate the file entity against the bound schema
    example_file.annotations = {"foo": "test_value", "bar": 43}  # Example annotations
    await example_file.store_async()
    print("Added annotations to file and stored")
    time.sleep(2)  # Allow time for processing

    validation_response_file = await example_file.validate_schema_async()
    print(f"File validation response: {validation_response_file}")

asyncio.run(main())

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

async def validate_schema_async(
    self, *, synapse_client: Optional["Synapse"] = None
) -> Union[JSONSchemaValidation, InvalidJSONSchemaValidation]:
    """
    Validate the entity against the bound JSON schema.

    Arguments:
        synapse_client (Optional[Synapse], optional): The Synapse client instance. If not provided,
            the last created instance from the Synapse class constructor will be used.

    Returns:
        The validation results.

    Example: Using this function
        Validating a folder or file against the bound JSON schema. This example demonstrates
        how to validate entities with annotations against their bound schemas. Requires entities
        to have schemas already bound. Set the `PROJECT_NAME` and `FILE_PATH` variables to your project name
        and file path respectively.

        ```python
        import asyncio
        import time
        from synapseclient import Synapse
        from synapseclient.models import File, Folder

        syn = Synapse()
        syn.login()

        # Define Project and JSON schema info
        PROJECT_NAME = "test_json_schema_project"  # replace with your project name
        FILE_PATH = "~/Sample.txt"  # replace with your test file path

        PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
        ORG_NAME = "UniqueOrg"  # replace with your organization name
        SCHEMA_NAME = "myTestSchema"  # replace with your schema name
        FOLDER_NAME = "test_script_folder"
        VERSION = "0.0.1"
        SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

        # Create organization (if not already created)
        js = syn.service("json_schema")
        all_orgs = js.list_organizations()
        for org in all_orgs:
            if org["name"] == ORG_NAME:
                print(f"Organization {ORG_NAME} already exists: {org}")
                break
        else:
            print(f"Creating organization {ORG_NAME}.")
            created_organization = js.create_organization(ORG_NAME)
            print(f"Created organization: {created_organization}")

        my_test_org = js.JsonSchemaOrganization(ORG_NAME)
        test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

        if not test_schema:
            # Create the schema (if not already created)
            schema_definition = {
                "$id": "mySchema",
                "type": "object",
                "properties": {
                    "foo": {"type": "string"},
                    "bar": {"type": "integer"},
                },
                "required": ["foo"]
            }
            test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
            print(f"Created new schema: {SCHEMA_NAME}")

        async def main():
            # Create a test folder
            test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
            await test_folder.store_async()
            print(f"Created test folder: {FOLDER_NAME}")

            # Bind JSON schema to the folder
            bound_schema = await test_folder.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Bound schema to folder: {bound_schema}")

            # Create and bind schema to a file
            example_file = File(
                path=FILE_PATH,  # Replace with your test file path
                parent_id=test_folder.id,
            )
            await example_file.store_async()

            bound_schema_file = await example_file.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Bound schema to file: {bound_schema_file}")

            # Validate the folder entity against the bound schema
            test_folder.annotations = {"foo": "test_value", "bar": 42}  # Example annotations
            await test_folder.store_async()
            print("Added annotations to folder and stored")
            time.sleep(2)  # Allow time for processing

            validation_response = await test_folder.validate_schema_async()
            print(f"Folder validation response: {validation_response}")

            # Validate the file entity against the bound schema
            example_file.annotations = {"foo": "test_value", "bar": 43}  # Example annotations
            await example_file.store_async()
            print("Added annotations to file and stored")
            time.sleep(2)  # Allow time for processing

            validation_response_file = await example_file.validate_schema_async()
            print(f"File validation response: {validation_response_file}")

        asyncio.run(main())
        ```
    """
    response = await validate_entity_with_json_schema(
        synapse_id=self.id, synapse_client=synapse_client
    )
    if "validationException" in response:
        return InvalidJSONSchemaValidation(
            validation_response=JSONSchemaValidation(
                object_id=response.get("objectId", None),
                object_type=response.get("objectType", None),
                object_etag=response.get("objectEtag", None),
                id=response.get("schema$id", None),
                is_valid=response.get("isValid", None),
                validated_on=response.get("validatedOn", None),
            ),
            validation_error_message=response.get("validationErrorMessage", None),
            all_validation_messages=response.get("allValidationMessages", []),
            validation_exception=ValidationException(
                pointer_to_violation=response.get("validationException", {}).get(
                    "pointerToViolation", None
                ),
                message=response.get("validationException", {}).get(
                    "message", None
                ),
                schema_location=response.get("validationException", {}).get(
                    "schemaLocation", None
                ),
                causing_exceptions=[
                    CausingException(
                        keyword=ce.get("keyword", None),
                        pointer_to_violation=ce.get("pointerToViolation", None),
                        message=ce.get("message", None),
                        schema_location=ce.get("schemaLocation", None),
                        causing_exceptions=[
                            CausingException(
                                keyword=nce.get("keyword", None),
                                pointer_to_violation=nce.get(
                                    "pointerToViolation", None
                                ),
                                message=nce.get("message", None),
                                schema_location=nce.get("schemaLocation", None),
                            )
                            for nce in ce.get("causingExceptions", [])
                        ],
                    )
                    for ce in response.get("validationException", {}).get(
                        "causingExceptions", []
                    )
                ],
            ),
        )
    return JSONSchemaValidation(
        object_id=response.get("objectId", None),
        object_type=response.get("objectType", None),
        object_etag=response.get("objectEtag", None),
        id=response.get("schema$id", None),
        is_valid=response.get("isValid", None),
        validated_on=response.get("validatedOn", None),
    )

get_schema_derived_keys_async async

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

Retrieve derived JSON schema keys for the entity.

PARAMETER	DESCRIPTION
synapse_client	The Synapse client instance. If not provided, the last created instance from the Synapse class constructor will be used. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
JSONSchemaDerivedKeys	An object containing the derived keys for the entity.

Using this function

Retrieving derived keys from a folder or file. This example demonstrates how to get derived annotation keys from schemas with constant values. Set the PROJECT_NAME variable to your project name.

import asyncio
from synapseclient import Synapse
from synapseclient.models import File, Folder

syn = Synapse()
syn.login()

# Define Project and JSON schema info
PROJECT_NAME = "test_json_schema_project"  # replace with your project name
FILE_PATH = "~/Sample.txt"  # replace with your test file path

PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
ORG_NAME = "UniqueOrg"  # replace with your organization name
DERIVED_TEST_SCHEMA_NAME = "myTestDerivedSchema"  # replace with your derived schema name
FOLDER_NAME = "test_script_folder"
VERSION = "0.0.1"
SCHEMA_URI = f"{ORG_NAME}-{DERIVED_TEST_SCHEMA_NAME}-{VERSION}"

# Create organization (if not already created)
js = syn.service("json_schema")
all_orgs = js.list_organizations()
for org in all_orgs:
    if org["name"] == ORG_NAME:
        print(f"Organization {ORG_NAME} already exists: {org}")
        break
else:
    print(f"Creating organization {ORG_NAME}.")
    created_organization = js.create_organization(ORG_NAME)
    print(f"Created organization: {created_organization}")

my_test_org = js.JsonSchemaOrganization(ORG_NAME)
test_schema = my_test_org.get_json_schema(DERIVED_TEST_SCHEMA_NAME)

if not test_schema:
    # Create the schema (if not already created)
    schema_definition = {
        "$id": "mySchema",
        "type": "object",
        "properties": {
            "foo": {"type": "string"},
            "baz": {"type": "string", "const": "example_value"},  # Example constant for derived annotation
            "bar": {"type": "integer"},
        },
        "required": ["foo"]
    }
    test_schema = my_test_org.create_json_schema(schema_definition, DERIVED_TEST_SCHEMA_NAME, VERSION)
    print(f"Created new derived schema: {DERIVED_TEST_SCHEMA_NAME}")

async def main():
    # Create a test folder
    test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
    await test_folder.store_async()
    print(f"Created test folder: {FOLDER_NAME}")

    # Bind JSON schema to the folder
    bound_schema = await test_folder.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Bound schema to folder with derived annotations: {bound_schema}")

    # Create and bind schema to a file
    example_file = File(
        path=FILE_PATH,  # Replace with your test file path
        parent_id=test_folder.id,
    )
    await example_file.store_async()

    bound_schema_file = await example_file.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Bound schema to file with derived annotations: {bound_schema_file}")

    # Get the derived keys from the bound schema of the folder
    test_folder.annotations = {"foo": "test_value_new", "bar": 42}  # Example annotations
    await test_folder.store_async()
    print("Added annotations to folder and stored")

    derived_keys = await test_folder.get_schema_derived_keys_async()
    print(f"Derived keys from folder: {derived_keys}")

    # Get the derived keys from the bound schema of the file
    example_file.annotations = {"foo": "test_value_new", "bar": 43}  # Example annotations
    await example_file.store_async()
    print("Added annotations to file and stored")

    derived_keys_file = await example_file.get_schema_derived_keys_async()
    print(f"Derived keys from file: {derived_keys_file}")

asyncio.run(main())

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

async def get_schema_derived_keys_async(
    self, *, synapse_client: Optional["Synapse"] = None
) -> JSONSchemaDerivedKeys:
    """
    Retrieve derived JSON schema keys for the entity.

    Arguments:
        synapse_client (Optional[Synapse], optional): The Synapse client instance. If not provided,
            the last created instance from the Synapse class constructor will be used.

    Returns:
        An object containing the derived keys for the entity.

    Example: Using this function
        Retrieving derived keys from a folder or file. This example demonstrates
        how to get derived annotation keys from schemas with constant values.
        Set the `PROJECT_NAME` variable to your project name.

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import File, Folder

        syn = Synapse()
        syn.login()

        # Define Project and JSON schema info
        PROJECT_NAME = "test_json_schema_project"  # replace with your project name
        FILE_PATH = "~/Sample.txt"  # replace with your test file path

        PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
        ORG_NAME = "UniqueOrg"  # replace with your organization name
        DERIVED_TEST_SCHEMA_NAME = "myTestDerivedSchema"  # replace with your derived schema name
        FOLDER_NAME = "test_script_folder"
        VERSION = "0.0.1"
        SCHEMA_URI = f"{ORG_NAME}-{DERIVED_TEST_SCHEMA_NAME}-{VERSION}"

        # Create organization (if not already created)
        js = syn.service("json_schema")
        all_orgs = js.list_organizations()
        for org in all_orgs:
            if org["name"] == ORG_NAME:
                print(f"Organization {ORG_NAME} already exists: {org}")
                break
        else:
            print(f"Creating organization {ORG_NAME}.")
            created_organization = js.create_organization(ORG_NAME)
            print(f"Created organization: {created_organization}")

        my_test_org = js.JsonSchemaOrganization(ORG_NAME)
        test_schema = my_test_org.get_json_schema(DERIVED_TEST_SCHEMA_NAME)

        if not test_schema:
            # Create the schema (if not already created)
            schema_definition = {
                "$id": "mySchema",
                "type": "object",
                "properties": {
                    "foo": {"type": "string"},
                    "baz": {"type": "string", "const": "example_value"},  # Example constant for derived annotation
                    "bar": {"type": "integer"},
                },
                "required": ["foo"]
            }
            test_schema = my_test_org.create_json_schema(schema_definition, DERIVED_TEST_SCHEMA_NAME, VERSION)
            print(f"Created new derived schema: {DERIVED_TEST_SCHEMA_NAME}")

        async def main():
            # Create a test folder
            test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
            await test_folder.store_async()
            print(f"Created test folder: {FOLDER_NAME}")

            # Bind JSON schema to the folder
            bound_schema = await test_folder.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Bound schema to folder with derived annotations: {bound_schema}")

            # Create and bind schema to a file
            example_file = File(
                path=FILE_PATH,  # Replace with your test file path
                parent_id=test_folder.id,
            )
            await example_file.store_async()

            bound_schema_file = await example_file.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Bound schema to file with derived annotations: {bound_schema_file}")

            # Get the derived keys from the bound schema of the folder
            test_folder.annotations = {"foo": "test_value_new", "bar": 42}  # Example annotations
            await test_folder.store_async()
            print("Added annotations to folder and stored")

            derived_keys = await test_folder.get_schema_derived_keys_async()
            print(f"Derived keys from folder: {derived_keys}")

            # Get the derived keys from the bound schema of the file
            example_file.annotations = {"foo": "test_value_new", "bar": 43}  # Example annotations
            await example_file.store_async()
            print("Added annotations to file and stored")

            derived_keys_file = await example_file.get_schema_derived_keys_async()
            print(f"Derived keys from file: {derived_keys_file}")

        asyncio.run(main())
        ```
    """
    response = await get_json_schema_derived_keys(
        synapse_id=self.id, synapse_client=synapse_client
    )
    return JSONSchemaDerivedKeys(keys=response["keys"])

---

synapseclient.models.RecordBasedMetadataTaskProperties dataclass

A CurationTaskProperties for record-based metadata.

Represents a Synapse RecordBasedMetadataTaskProperties.

ATTRIBUTE	DESCRIPTION
record_set_id	The synId of the RecordSet that will contain all record-based metadata TYPE: Optional[str]

Source code in synapseclient/models/curation.py

@dataclass
class RecordBasedMetadataTaskProperties:
    """
    A CurationTaskProperties for record-based metadata.

    Represents a [Synapse RecordBasedMetadataTaskProperties](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/curation/metadata/RecordBasedMetadataTaskProperties.html).

    Attributes:
        record_set_id: The synId of the RecordSet that will contain all record-based metadata
    """

    record_set_id: Optional[str] = None
    """The synId of the RecordSet that will contain all record-based metadata"""

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

        Arguments:
            synapse_response: The response from the REST API.

        Returns:
            The RecordBasedMetadataTaskProperties object.
        """
        self.record_set_id = synapse_response.get("recordSetId", None)
        return self

    def to_synapse_request(self) -> Dict[str, Any]:
        """
        Converts this dataclass to a dictionary suitable for a Synapse REST API request.

        Returns:
            A dictionary representation of this object for API requests.
        """
        request_dict = {"concreteType": RECORD_BASED_METADATA_TASK_PROPERTIES}
        if self.record_set_id is not None:
            request_dict["recordSetId"] = self.record_set_id
        return request_dict

Attributes

record_set_id class-attribute instance-attribute

record_set_id: Optional[str] = None

The synId of the RecordSet that will contain all record-based metadata

---

synapseclient.models.FileBasedMetadataTaskProperties dataclass

A CurationTaskProperties for file-based data, describing where data is uploaded and a view which contains the annotations.

Represents a Synapse FileBasedMetadataTaskProperties.

ATTRIBUTE	DESCRIPTION
upload_folder_id	The synId of the folder where data files of this type are to be uploaded TYPE: Optional[str]
file_view_id	The synId of the FileView that shows all data of this type TYPE: Optional[str]

Source code in synapseclient/models/curation.py

@dataclass
class FileBasedMetadataTaskProperties:
    """
    A CurationTaskProperties for file-based data, describing where data is uploaded
    and a view which contains the annotations.

    Represents a [Synapse FileBasedMetadataTaskProperties](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/curation/metadata/FileBasedMetadataTaskProperties.html).

    Attributes:
        upload_folder_id: The synId of the folder where data files of this type are to be uploaded
        file_view_id: The synId of the FileView that shows all data of this type
    """

    upload_folder_id: Optional[str] = None
    """The synId of the folder where data files of this type are to be uploaded"""

    file_view_id: Optional[str] = None
    """The synId of the FileView that shows all data of this type"""

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

        Arguments:
            synapse_response: The response from the REST API.

        Returns:
            The FileBasedMetadataTaskProperties object.
        """
        self.upload_folder_id = synapse_response.get("uploadFolderId", None)
        self.file_view_id = synapse_response.get("fileViewId", None)
        return self

    def to_synapse_request(self) -> Dict[str, Any]:
        """
        Converts this dataclass to a dictionary suitable for a Synapse REST API request.

        Returns:
            A dictionary representation of this object for API requests.
        """
        request_dict = {"concreteType": FILE_BASED_METADATA_TASK_PROPERTIES}
        if self.upload_folder_id is not None:
            request_dict["uploadFolderId"] = self.upload_folder_id
        if self.file_view_id is not None:
            request_dict["fileViewId"] = self.file_view_id
        return request_dict

Attributes

upload_folder_id class-attribute instance-attribute

upload_folder_id: Optional[str] = None

The synId of the folder where data files of this type are to be uploaded

file_view_id class-attribute instance-attribute

file_view_id: Optional[str] = None

The synId of the FileView that shows all data of this type

---

synapseclient.models.Grid dataclass

Bases: GridSynchronousProtocol

A GridSession provides functionality to create and manage grid sessions in Synapse. Grid sessions are used for curation workflows where data can be edited in a grid format and then exported back to record sets.

ATTRIBUTE	DESCRIPTION
record_set_id	The synId of the RecordSet to use for initializing the grid TYPE: Optional[str]
initial_query	Initialize a grid session from an EntityView. Mutually exclusive with record_set_id. TYPE: Optional[Query]
owner_principal_id	The principal ID (user or team) that will own the created grid session. When not provided, the principal ID of the caller is used. TYPE: int | None
session_id	The unique sessionId that identifies the grid session TYPE: Optional[str]
started_by	The user that started this session TYPE: Optional[str]
started_on	The date-time when the session was started TYPE: Optional[str]
etag	Changes when the session changes TYPE: Optional[str]
modified_on	The date-time when the session was last changed TYPE: Optional[str]
last_replica_id_client	The last replica ID issued to a client TYPE: Optional[int]
last_replica_id_service	The last replica ID issued to a service TYPE: Optional[int]
grid_json_schema_id	The $id of the JSON schema used for model validation TYPE: Optional[str]
source_entity_id	The synId of the table/view/csv that this grid was cloned from TYPE: Optional[str]
record_set_version_number	The version number of the exported record set TYPE: Optional[int]
validation_summary_statistics	Summary statistics for validation results TYPE: Optional[ValidationSummary]

Create and manage a grid session workflow

 

from synapseclient import Synapse
from synapseclient.models import Grid

syn = Synapse()
syn.login()

# Create a new grid session from a record set
grid = Grid(record_set_id="syn1234567")
grid = grid.create()
print(f"Created grid session: {grid.session_id}")

# Later, export the modified data back to the record set
grid = grid.export_to_record_set()
print(f"Exported to version: {grid.record_set_version_number}")

# Clean up by deleting the session when done
grid.delete()

Working with grid sessions using queries

 

from synapseclient import Synapse
from synapseclient.models import Grid
from synapseclient.models.table_components import Query

syn = Synapse()
syn.login()

# Create a grid from an entity view query
query = Query(sql="SELECT * FROM syn1234567")
grid = Grid(initial_query=query)
grid = grid.create()

# Work with the grid session...
# Export when ready
grid = grid.export_to_record_set()

Source code in synapseclient/models/curation.py

@dataclass
@async_to_sync
class Grid(GridSynchronousProtocol):
    """
    A GridSession provides functionality to create and manage grid sessions in Synapse.
    Grid sessions are used for curation workflows where data can be edited in a grid format
    and then exported back to record sets.

    Attributes:
        record_set_id: The synId of the RecordSet to use for initializing the grid
        initial_query: Initialize a grid session from an EntityView.
            Mutually exclusive with record_set_id.
        owner_principal_id: The principal ID (user or team) that will own the
            created grid session. When not provided, the principal ID of the
            caller is used.
        session_id: The unique sessionId that identifies the grid session
        started_by: The user that started this session
        started_on: The date-time when the session was started
        etag: Changes when the session changes
        modified_on: The date-time when the session was last changed
        last_replica_id_client: The last replica ID issued to a client
        last_replica_id_service: The last replica ID issued to a service
        grid_json_schema_id: The $id of the JSON schema used for model validation
        source_entity_id: The synId of the table/view/csv that this grid was cloned from
        record_set_version_number: The version number of the exported record set
        validation_summary_statistics: Summary statistics for validation results

    Example: Create and manage a grid session workflow
         

        ```python
        from synapseclient import Synapse
        from synapseclient.models import Grid

        syn = Synapse()
        syn.login()

        # Create a new grid session from a record set
        grid = Grid(record_set_id="syn1234567")
        grid = grid.create()
        print(f"Created grid session: {grid.session_id}")

        # Later, export the modified data back to the record set
        grid = grid.export_to_record_set()
        print(f"Exported to version: {grid.record_set_version_number}")

        # Clean up by deleting the session when done
        grid.delete()
        ```

    Example: Working with grid sessions using queries
         

        ```python
        from synapseclient import Synapse
        from synapseclient.models import Grid
        from synapseclient.models.table_components import Query

        syn = Synapse()
        syn.login()

        # Create a grid from an entity view query
        query = Query(sql="SELECT * FROM syn1234567")
        grid = Grid(initial_query=query)
        grid = grid.create()

        # Work with the grid session...
        # Export when ready
        grid = grid.export_to_record_set()
        ```
    """

    record_set_id: Optional[str] = None
    """The synId of the RecordSet to use for initializing the grid"""

    initial_query: Optional[Query] = None
    """Initialize a grid session from an EntityView.
    Mutually exclusive with record_set_id."""

    owner_principal_id: int | None = None
    """The principal ID (user or team) that will own the created grid session.
    When not provided, the principal ID of the caller is used."""

    session_id: Optional[str] = None
    """The unique sessionId that identifies the grid session"""

    started_by: Optional[str] = None
    """The user that started this session"""

    started_on: Optional[str] = None
    """The date-time when the session was started"""

    etag: Optional[str] = None
    """Changes when the session changes"""

    modified_on: Optional[str] = None
    """The date-time when the session was last changed"""

    last_replica_id_client: Optional[int] = None
    """The last replica ID issued to a client. Client replica IDs are incremented."""

    last_replica_id_service: Optional[int] = None
    """The last replica ID issued to a service. Service replica IDs are decremented."""

    grid_json_schema_id: Optional[str] = None
    """The $id of the JSON schema that will be used for model validation in this grid session"""

    source_entity_id: Optional[str] = None
    """The synId of the table/view/csv that this grid was cloned from"""

    record_set_version_number: Optional[int] = None
    """The version number of the exported record set"""

    validation_summary_statistics: Optional[ValidationSummary] = None
    """Summary statistics for validation results"""

    async def create_async(
        self,
        attach_to_previous_session=False,
        *,
        timeout: int = 120,
        synapse_client: Optional[Synapse] = None,
    ) -> "Grid":
        """
        Creates a new grid session from a `record_set_id` or `initial_query`.

        When using `record_set_id`, first checks for existing active sessions that match
        the record set before creating a new one. When using `initial_query`, always
        creates a new session due to the complexity of matching query parameters.

        Arguments:
            attach_to_previous_session: If True and using `record_set_id`, will attach
                to an existing active session if one exists. Defaults to False.
            timeout: The number of seconds to wait for the job to complete or progress
                before raising a SynapseTimeoutError. Defaults to 120.
            synapse_client: If not passed 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:
            GridSession: The GridSession object with populated session_id.

        Raises:
            ValueError: If `record_set_id` or `initial_query` is not provided.

        Example: Create a grid session asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import Grid

            syn = Synapse()
            syn.login()

            async def main():
                # Create a grid session from a record set
                grid = Grid(record_set_id="syn1234567")
                grid = await grid.create_async()
                print(f"Created grid session: {grid.session_id}")

            asyncio.run(main())
            ```
        """
        if not self.record_set_id and not self.initial_query:
            raise ValueError(
                "record_set_id or initial_query is required to create a GridSession"
            )

        trace.get_current_span().set_attributes(
            {
                "synapse.record_set_id": self.record_set_id or "",
                "synapse.session_id": self.session_id or "",
            }
        )

        # Check for existing active sessions only when using record_set_id
        # For initial_query, always create a new session due to complexity of matching
        if self.record_set_id and attach_to_previous_session:
            # Look for existing active sessions for this record set
            async for existing_session in self.list_async(
                source_id=self.record_set_id, synapse_client=synapse_client
            ):
                # Found an existing session, populate this object with its data and return
                self.session_id = existing_session.session_id
                self.started_by = existing_session.started_by
                self.started_on = existing_session.started_on
                self.etag = existing_session.etag
                self.modified_on = existing_session.modified_on
                self.last_replica_id_client = existing_session.last_replica_id_client
                self.last_replica_id_service = existing_session.last_replica_id_service
                self.grid_json_schema_id = existing_session.grid_json_schema_id
                self.source_entity_id = existing_session.source_entity_id
                return self

        # No existing session found, create a new one
        create_request = CreateGridRequest(
            record_set_id=self.record_set_id,
            initial_query=self.initial_query,
            owner_principal_id=self.owner_principal_id,
        )
        result = await create_request.send_job_and_wait_async(
            timeout=timeout, synapse_client=synapse_client
        )

        # Fill this GridSession with the grid session data from the async job response
        result.fill_grid_session_from_response(self)

        return self

    async def export_to_record_set_async(
        self, *, timeout: int = 120, synapse_client: Optional[Synapse] = None
    ) -> "Grid":
        """
        Exports the grid session data back to a record set. This will create a new version
        of the original record set with the modified data from the grid session.

        Arguments:
            timeout: The number of seconds to wait for the job to complete or progress
                before raising a SynapseTimeoutError. Defaults to 120.
            synapse_client: If not passed 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:
            GridSession: The GridSession object with export information populated.

        Raises:
            ValueError: If session_id is not provided.

        Example: Export grid session data back to record set asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import Grid

            syn = Synapse()
            syn.login()

            async def main():
                # Export modified grid data back to the record set
                grid = Grid(session_id="abc-123-def")
                grid = await grid.export_to_record_set_async()
                print(f"Exported to record set: {grid.record_set_id}")
                print(f"Version number: {grid.record_set_version_number}")
                if grid.validation_summary_statistics:
                    print(f"Valid records: {grid.validation_summary_statistics.number_of_valid_children}")

            asyncio.run(main())
            ```
        """
        if not self.session_id:
            raise ValueError("session_id is required to export a GridSession")

        trace.get_current_span().set_attributes(
            {
                "synapse.session_id": self.session_id or "",
            }
        )

        # Create and send the export request
        export_request = GridRecordSetExportRequest(session_id=self.session_id)
        result = await export_request.send_job_and_wait_async(
            timeout=timeout, synapse_client=synapse_client
        )

        self.record_set_id = result.response_record_set_id
        self.record_set_version_number = result.record_set_version_number
        self.validation_summary_statistics = result.validation_summary_statistics

        return self

    def fill_from_dict(self, synapse_response: Dict[str, Any]) -> "Grid":
        """Converts a response from the REST API into this dataclass."""
        self.session_id = synapse_response.get("sessionId", None)
        self.started_by = synapse_response.get("startedBy", None)
        self.started_on = synapse_response.get("startedOn", None)
        self.etag = synapse_response.get("etag", None)
        self.modified_on = synapse_response.get("modifiedOn", None)
        self.last_replica_id_client = synapse_response.get("lastReplicaIdClient", None)
        self.last_replica_id_service = synapse_response.get(
            "lastReplicaIdService", None
        )
        self.grid_json_schema_id = synapse_response.get("gridJsonSchema$Id", None)
        self.source_entity_id = synapse_response.get("sourceEntityId", None)
        self.owner_principal_id = synapse_response.get("ownerPrincipalId")
        return self

    @skip_async_to_sync
    @classmethod
    async def list_async(
        cls,
        source_id: Optional[str] = None,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> AsyncGenerator["Grid", None]:
        """
        Generator to get a list of active grid sessions for the user.

        Arguments:
            source_id: Optional. When provided, only sessions with this synId will be returned.
            synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Yields:
            Grid objects representing active grid sessions.

        Example: List all active grid sessions asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import Grid

            syn = Synapse()
            syn.login()

            async def main():
                # List all active grid sessions for the user
                async for grid in Grid.list_async():
                    print(f"Session ID: {grid.session_id}")
                    print(f"Source Entity: {grid.source_entity_id}")
                    print(f"Started: {grid.started_on}")
                    print("---")

                # List grid sessions for a specific source
                async for grid in Grid.list_async(source_id="syn1234567"):
                    print(f"Session ID: {grid.session_id}")
                    print(f"Modified: {grid.modified_on}")

            asyncio.run(main())
            ```
        """
        async for session_dict in list_grid_sessions(
            source_id=source_id, synapse_client=synapse_client
        ):
            # Convert the dictionary to a Grid object
            grid = cls()
            grid.fill_from_dict(session_dict)
            yield grid

    @classmethod
    def list(
        cls,
        source_id: Optional[str] = None,
        *,
        synapse_client: Optional[Synapse] = None,
    ) -> Generator["Grid", None, None]:
        """
        Generator to get a list of active grid sessions for the user.

        Arguments:
            source_id: Optional. When provided, only sessions with this synId will be returned.
            synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

        Yields:
            Grid objects representing active grid sessions.
        """
        return wrap_async_generator_to_sync_generator(
            async_gen_func=cls.list_async,
            source_id=source_id,
            synapse_client=synapse_client,
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Grid_Delete: ID: {self.session_id}"
    )
    async def delete_async(self, *, synapse_client: Optional[Synapse] = None) -> None:
        """
        Delete the grid session.

        Note: Only the user that created a grid session may delete it.

        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 session_id is not provided.

        Example: Delete a grid session asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import Grid

            syn = Synapse()
            syn.login()

            async def main():
                # Delete the grid session
                grid = Grid(session_id="abc-123-def")
                await grid.delete_async()
                print("Grid session deleted successfully")

            asyncio.run(main())
            ```
        """
        if not self.session_id:
            raise ValueError("session_id is required to delete a GridSession")

        trace.get_current_span().set_attributes(
            {
                "synapse.session_id": self.session_id or "",
            }
        )

        await delete_grid_session(
            session_id=self.session_id, synapse_client=synapse_client
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Grid_ImportCsv: ID: {self.session_id}"
    )
    async def import_csv_async(
        self,
        path: str,
        *,
        timeout: int = 120,
        csv_table_descriptor: Optional[CsvTableDescriptor] = None,
        synapse_client: Optional[Synapse] = None,
    ) -> "Grid":
        """
        Import a CSV file into this grid session. Previews the file to determine
        the column schema, then imports the data. Currently supports only grids
        created from a record set.

        Arguments:
            path: Local path to the CSV file to import.
            csv_table_descriptor: The description of the CSV format (delimiter,
                quote character, etc.). If not provided, the default CSV format
                will be used.
            timeout: The number of seconds to wait for each async job to complete
                or progress before raising a SynapseTimeoutError. Defaults to 120.
            synapse_client: If not passed 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 Grid object.

        Raises:
            ValueError: If session_id is not provided.

        Example: Import a CSV file into a grid session asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import Grid

            syn = Synapse()
            syn.login()

            async def main():
                grid = Grid(session_id="abc-123-def")
                grid = await grid.import_csv_async(path="/local/path/to/data.csv")
                print(f"Import complete for session: {grid.session_id}")

            asyncio.run(main())
            ```
        """

        if not self.session_id:
            raise ValueError(
                "session_id is required to import a CSV into a GridSession"
            )

        if not os.path.isfile(path):
            raise ValueError(f"Path '{path}' is not a valid file.")

        trace.get_current_span().set_attributes(
            {
                "synapse.session_id": self.session_id,
            }
        )

        client = Synapse.get_client(synapse_client=synapse_client)
        file_handle = await upload_synapse_s3(syn=client, file_path=path)
        file_handle_id = file_handle["id"]

        effective_descriptor = csv_table_descriptor or CsvTableDescriptor()

        upload_to_table_preview = UploadToTablePreviewRequest(
            csv_table_descriptor=effective_descriptor,
            upload_file_handle_id=file_handle_id,
        )

        preview_response = await upload_to_table_preview.send_job_and_wait_async(
            timeout=timeout, synapse_client=synapse_client
        )
        if not preview_response.suggested_columns:
            raise ValueError(
                f"CSV preview for file handle {file_handle_id} returned no suggested "
                f"columns (rows scanned: {preview_response.rows_scanned}). The file may "
                f"be empty, contain only a header row, or use a separator different "
                f"from the configured csv_table_descriptor "
                f"(separator={repr(effective_descriptor.separator)})."
            )

        import_request = GridCsvImportRequest(
            session_id=self.session_id,
            file_handle_id=file_handle_id,
            schema=preview_response.suggested_columns,
            csv_descriptor=effective_descriptor,
        )
        import_response = await import_request.send_job_and_wait_async(
            timeout=timeout, synapse_client=synapse_client
        )
        client.logger.info(
            f"CSV import to grid session {self.session_id} completed successfully, "
            f"total count: {import_response.total_count}, "
            f"total created: {import_response.created_count}, "
            f"total updated: {import_response.updated_count}"
        )

        return self

    @otel_trace_method(
        method_to_trace_name=lambda self, *args, **kwargs: f"Grid_DownloadCsv: ID: {self.session_id}"
    )
    async def download_csv_async(
        self,
        *,
        destination: Optional[str] = None,
        write_header: bool = True,
        include_row_id_and_row_version: bool = False,
        include_etag: bool = False,
        csv_table_descriptor: Optional[CsvTableDescriptor] = None,
        file_name: Optional[str] = None,
        timeout: int = 120,
        synapse_client: Optional[Synapse] = None,
    ) -> str:
        """
        Asynchronously download the current state of this grid session as a CSV file.

        Submits a DownloadFromGridRequest async job, waits for it to complete,
        then downloads the resulting CSV to the local filesystem.

        Arguments:
            destination: Local directory path where the CSV will be saved. The directory must already exist.
                If not provided, defaults to the current working directory.
            write_header: Whether the first line should contain column names
                as a header. Defaults to True.
            include_row_id_and_row_version: Whether the first two columns
                should contain row ID and version. Defaults to False.
            include_etag: Whether a column should contain the row etag.
                Defaults to False.
            csv_table_descriptor: The description of the CSV format (delimiter,
                quote character, etc.). If not provided, the default CSV format
                will be used.
            file_name: The optional name for the downloaded file. If not
                provided, defaults to `grid_{session_id}-{timestamp}.csv`.
            timeout: The number of seconds to wait for the async job to
                complete or progress before raising a SynapseTimeoutError.
                Defaults to 120.
            synapse_client: If not passed 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 local path to the downloaded CSV file.

        Raises:
            ValueError: If session_id is not provided.

        Example: Download a grid session as a CSV asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import Grid

            syn = Synapse()
            syn.login()

            async def main():
                grid = Grid(session_id="abc-123-def")
                path = await grid.download_csv_async(destination="./downloads")
                print(f"Downloaded CSV to: {path}")

            asyncio.run(main())
            ```

        Example: Download a grid session as a CSV with a custom file name asynchronously
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import Grid

            syn = Synapse()
            syn.login()

            async def main():
                grid = Grid(session_id="abc-123-def")
                path = await grid.download_csv_async(
                    destination="./downloads", file_name="my_export.csv"
                )
                print(f"Downloaded CSV to: {path}")

            asyncio.run(main())
            ```
        """
        if not self.session_id:
            raise ValueError("session_id is required to download a GridSession as CSV")

        if not destination:
            destination = os.getcwd()

        if not os.path.isdir(destination):
            raise ValueError(f"Destination {destination} is not a valid directory.")

        trace.get_current_span().set_attributes({"synapse.session_id": self.session_id})

        effective_descriptor = csv_table_descriptor or CsvTableDescriptor()
        request = DownloadFromGridRequest(
            session_id=self.session_id,
            write_header=write_header,
            include_row_id_and_row_version=include_row_id_and_row_version,
            include_etag=include_etag,
            csv_table_descriptor=effective_descriptor,
            file_name=file_name,
        )
        download_response = await request.send_job_and_wait_async(
            timeout=timeout, synapse_client=synapse_client
        )
        if not download_response.results_file_handle_id:
            raise ValueError(
                f"Download job for grid session '{self.session_id}' completed but "
                "did not return a file handle ID. The CSV result may be empty or "
                "the job may have failed silently."
            )
        file_handle, presigned_url = await asyncio.gather(
            get_file_handle(
                file_handle_id=download_response.results_file_handle_id,
                synapse_client=synapse_client,
            ),
            get_file_handle_presigned_url(
                file_handle_id=download_response.results_file_handle_id,
                synapse_client=synapse_client,
            ),
        )
        if not file_name:
            timestamp = datetime.now(tz=timezone.utc).strftime("%Y%m%d%H%M%S")
            file_name = f"grid_{self.session_id}-{timestamp}.csv"
        file_path = os.path.join(destination, file_name)
        return await asyncio.to_thread(
            download_from_url,
            url=presigned_url,
            destination=file_path,
            file_handle_id=file_handle["id"],
            expected_md5=file_handle.get("contentMd5"),
            url_is_presigned=True,
            synapse_client=synapse_client,
        )

    @otel_trace_method(
        method_to_trace_name=lambda self, **kwargs: f"Grid_Synchronize: ID: {self.session_id}"
    )
    async def synchronize_async(
        self, *, timeout: int = 120, synapse_client: Optional[Synapse] = None
    ) -> "Grid":
        """
        Synchronizes the grid session's schema and row data against its source entity.

        This is intended for grid sessions created from a file view via `initial_query`.
        Grid sessions backed by a RecordSet should use `export_to_record_set` instead.

        Arguments:
            timeout: The number of seconds to wait for the job to complete or progress
                before raising a SynapseTimeoutError. Defaults to 120.
            synapse_client: If not passed 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:
            Grid: The Grid object.

        Raises:
            ValueError: If session_id is not provided.

        Example: Synchronize a grid session created from a file view
             

            ```python
            import asyncio
            from synapseclient import Synapse
            from synapseclient.models import Grid
            from synapseclient.models.table_components import Query

            syn = Synapse()
            syn.login()

            async def main():
                # First create a grid session from a file view
                query = Query(sql="SELECT * FROM syn1234567")
                grid = Grid(initial_query=query)
                grid = await grid.create_async()

                # Synchronize the grid with the latest state of the file view
                grid = await grid.synchronize_async()

            asyncio.run(main())
            ```
        """
        if not self.session_id:
            raise ValueError("session_id is required to synchronize a GridSession")

        request = SynchronizeGridRequest(grid_session_id=self.session_id)
        result = await request.send_job_and_wait_async(
            timeout=timeout, synapse_client=synapse_client
        )

        if result.error_messages:
            client = Synapse.get_client(synapse_client=synapse_client)
            client.logger.error(
                f"Grid session '{self.session_id}' synchronization completed with "
                f"error messages: {result.error_messages}"
            )

        return self

Methods:

create_async async

create_async(attach_to_previous_session=False, *, timeout: int = 120, synapse_client: Optional[Synapse] = None) -> Grid

Creates a new grid session from a record_set_id or initial_query.

When using record_set_id, first checks for existing active sessions that match the record set before creating a new one. When using initial_query, always creates a new session due to the complexity of matching query parameters.

PARAMETER	DESCRIPTION
attach_to_previous_session	If True and using record_set_id, will attach to an existing active session if one exists. Defaults to False. DEFAULT: False
timeout	The number of seconds to wait for the job to complete or progress before raising a SynapseTimeoutError. Defaults to 120. TYPE: int DEFAULT: 120
synapse_client	If not passed 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
GridSession	The GridSession object with populated session_id. TYPE: Grid

RAISES	DESCRIPTION
ValueError	If record_set_id or initial_query is not provided.

Create a grid session asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import Grid

syn = Synapse()
syn.login()

async def main():
    # Create a grid session from a record set
    grid = Grid(record_set_id="syn1234567")
    grid = await grid.create_async()
    print(f"Created grid session: {grid.session_id}")

asyncio.run(main())

Source code in synapseclient/models/curation.py

async def create_async(
    self,
    attach_to_previous_session=False,
    *,
    timeout: int = 120,
    synapse_client: Optional[Synapse] = None,
) -> "Grid":
    """
    Creates a new grid session from a `record_set_id` or `initial_query`.

    When using `record_set_id`, first checks for existing active sessions that match
    the record set before creating a new one. When using `initial_query`, always
    creates a new session due to the complexity of matching query parameters.

    Arguments:
        attach_to_previous_session: If True and using `record_set_id`, will attach
            to an existing active session if one exists. Defaults to False.
        timeout: The number of seconds to wait for the job to complete or progress
            before raising a SynapseTimeoutError. Defaults to 120.
        synapse_client: If not passed 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:
        GridSession: The GridSession object with populated session_id.

    Raises:
        ValueError: If `record_set_id` or `initial_query` is not provided.

    Example: Create a grid session asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import Grid

        syn = Synapse()
        syn.login()

        async def main():
            # Create a grid session from a record set
            grid = Grid(record_set_id="syn1234567")
            grid = await grid.create_async()
            print(f"Created grid session: {grid.session_id}")

        asyncio.run(main())
        ```
    """
    if not self.record_set_id and not self.initial_query:
        raise ValueError(
            "record_set_id or initial_query is required to create a GridSession"
        )

    trace.get_current_span().set_attributes(
        {
            "synapse.record_set_id": self.record_set_id or "",
            "synapse.session_id": self.session_id or "",
        }
    )

    # Check for existing active sessions only when using record_set_id
    # For initial_query, always create a new session due to complexity of matching
    if self.record_set_id and attach_to_previous_session:
        # Look for existing active sessions for this record set
        async for existing_session in self.list_async(
            source_id=self.record_set_id, synapse_client=synapse_client
        ):
            # Found an existing session, populate this object with its data and return
            self.session_id = existing_session.session_id
            self.started_by = existing_session.started_by
            self.started_on = existing_session.started_on
            self.etag = existing_session.etag
            self.modified_on = existing_session.modified_on
            self.last_replica_id_client = existing_session.last_replica_id_client
            self.last_replica_id_service = existing_session.last_replica_id_service
            self.grid_json_schema_id = existing_session.grid_json_schema_id
            self.source_entity_id = existing_session.source_entity_id
            return self

    # No existing session found, create a new one
    create_request = CreateGridRequest(
        record_set_id=self.record_set_id,
        initial_query=self.initial_query,
        owner_principal_id=self.owner_principal_id,
    )
    result = await create_request.send_job_and_wait_async(
        timeout=timeout, synapse_client=synapse_client
    )

    # Fill this GridSession with the grid session data from the async job response
    result.fill_grid_session_from_response(self)

    return self

export_to_record_set_async async

export_to_record_set_async(*, timeout: int = 120, synapse_client: Optional[Synapse] = None) -> Grid

Exports the grid session data back to a record set. This will create a new version of the original record set with the modified data from the grid session.

PARAMETER	DESCRIPTION
timeout	The number of seconds to wait for the job to complete or progress before raising a SynapseTimeoutError. Defaults to 120. TYPE: int DEFAULT: 120
synapse_client	If not passed 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
GridSession	The GridSession object with export information populated. TYPE: Grid

RAISES	DESCRIPTION
ValueError	If session_id is not provided.

Export grid session data back to record set asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import Grid

syn = Synapse()
syn.login()

async def main():
    # Export modified grid data back to the record set
    grid = Grid(session_id="abc-123-def")
    grid = await grid.export_to_record_set_async()
    print(f"Exported to record set: {grid.record_set_id}")
    print(f"Version number: {grid.record_set_version_number}")
    if grid.validation_summary_statistics:
        print(f"Valid records: {grid.validation_summary_statistics.number_of_valid_children}")

asyncio.run(main())

Source code in synapseclient/models/curation.py

async def export_to_record_set_async(
    self, *, timeout: int = 120, synapse_client: Optional[Synapse] = None
) -> "Grid":
    """
    Exports the grid session data back to a record set. This will create a new version
    of the original record set with the modified data from the grid session.

    Arguments:
        timeout: The number of seconds to wait for the job to complete or progress
            before raising a SynapseTimeoutError. Defaults to 120.
        synapse_client: If not passed 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:
        GridSession: The GridSession object with export information populated.

    Raises:
        ValueError: If session_id is not provided.

    Example: Export grid session data back to record set asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import Grid

        syn = Synapse()
        syn.login()

        async def main():
            # Export modified grid data back to the record set
            grid = Grid(session_id="abc-123-def")
            grid = await grid.export_to_record_set_async()
            print(f"Exported to record set: {grid.record_set_id}")
            print(f"Version number: {grid.record_set_version_number}")
            if grid.validation_summary_statistics:
                print(f"Valid records: {grid.validation_summary_statistics.number_of_valid_children}")

        asyncio.run(main())
        ```
    """
    if not self.session_id:
        raise ValueError("session_id is required to export a GridSession")

    trace.get_current_span().set_attributes(
        {
            "synapse.session_id": self.session_id or "",
        }
    )

    # Create and send the export request
    export_request = GridRecordSetExportRequest(session_id=self.session_id)
    result = await export_request.send_job_and_wait_async(
        timeout=timeout, synapse_client=synapse_client
    )

    self.record_set_id = result.response_record_set_id
    self.record_set_version_number = result.record_set_version_number
    self.validation_summary_statistics = result.validation_summary_statistics

    return self

synchronize_async async

synchronize_async(*, timeout: int = 120, synapse_client: Optional[Synapse] = None) -> Grid

Synchronizes the grid session's schema and row data against its source entity.

This is intended for grid sessions created from a file view via initial_query. Grid sessions backed by a RecordSet should use export_to_record_set instead.

PARAMETER	DESCRIPTION
timeout	The number of seconds to wait for the job to complete or progress before raising a SynapseTimeoutError. Defaults to 120. TYPE: int DEFAULT: 120
synapse_client	If not passed 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
Grid	The Grid object. TYPE: Grid

RAISES	DESCRIPTION
ValueError	If session_id is not provided.

Synchronize a grid session created from a file view

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import Grid
from synapseclient.models.table_components import Query

syn = Synapse()
syn.login()

async def main():
    # First create a grid session from a file view
    query = Query(sql="SELECT * FROM syn1234567")
    grid = Grid(initial_query=query)
    grid = await grid.create_async()

    # Synchronize the grid with the latest state of the file view
    grid = await grid.synchronize_async()

asyncio.run(main())

Source code in synapseclient/models/curation.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Grid_Synchronize: ID: {self.session_id}"
)
async def synchronize_async(
    self, *, timeout: int = 120, synapse_client: Optional[Synapse] = None
) -> "Grid":
    """
    Synchronizes the grid session's schema and row data against its source entity.

    This is intended for grid sessions created from a file view via `initial_query`.
    Grid sessions backed by a RecordSet should use `export_to_record_set` instead.

    Arguments:
        timeout: The number of seconds to wait for the job to complete or progress
            before raising a SynapseTimeoutError. Defaults to 120.
        synapse_client: If not passed 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:
        Grid: The Grid object.

    Raises:
        ValueError: If session_id is not provided.

    Example: Synchronize a grid session created from a file view
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import Grid
        from synapseclient.models.table_components import Query

        syn = Synapse()
        syn.login()

        async def main():
            # First create a grid session from a file view
            query = Query(sql="SELECT * FROM syn1234567")
            grid = Grid(initial_query=query)
            grid = await grid.create_async()

            # Synchronize the grid with the latest state of the file view
            grid = await grid.synchronize_async()

        asyncio.run(main())
        ```
    """
    if not self.session_id:
        raise ValueError("session_id is required to synchronize a GridSession")

    request = SynchronizeGridRequest(grid_session_id=self.session_id)
    result = await request.send_job_and_wait_async(
        timeout=timeout, synapse_client=synapse_client
    )

    if result.error_messages:
        client = Synapse.get_client(synapse_client=synapse_client)
        client.logger.error(
            f"Grid session '{self.session_id}' synchronization completed with "
            f"error messages: {result.error_messages}"
        )

    return self

download_csv_async async

download_csv_async(*, destination: Optional[str] = None, write_header: bool = True, include_row_id_and_row_version: bool = False, include_etag: bool = False, csv_table_descriptor: Optional[CsvTableDescriptor] = None, file_name: Optional[str] = None, timeout: int = 120, synapse_client: Optional[Synapse] = None) -> str

Asynchronously download the current state of this grid session as a CSV file.

Submits a DownloadFromGridRequest async job, waits for it to complete, then downloads the resulting CSV to the local filesystem.

PARAMETER	DESCRIPTION
destination	Local directory path where the CSV will be saved. The directory must already exist. If not provided, defaults to the current working directory. TYPE: Optional[str] DEFAULT: None
write_header	Whether the first line should contain column names as a header. Defaults to True. TYPE: bool DEFAULT: True
include_row_id_and_row_version	Whether the first two columns should contain row ID and version. Defaults to False. TYPE: bool DEFAULT: False
include_etag	Whether a column should contain the row etag. Defaults to False. TYPE: bool DEFAULT: False
csv_table_descriptor	The description of the CSV format (delimiter, quote character, etc.). If not provided, the default CSV format will be used. TYPE: Optional[CsvTableDescriptor] DEFAULT: None
file_name	The optional name for the downloaded file. If not provided, defaults to grid_{session_id}-{timestamp}.csv. TYPE: Optional[str] DEFAULT: None
timeout	The number of seconds to wait for the async job to complete or progress before raising a SynapseTimeoutError. Defaults to 120. TYPE: int DEFAULT: 120
synapse_client	If not passed 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 local path to the downloaded CSV file.

RAISES	DESCRIPTION
ValueError	If session_id is not provided.

Download a grid session as a CSV asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import Grid

syn = Synapse()
syn.login()

async def main():
    grid = Grid(session_id="abc-123-def")
    path = await grid.download_csv_async(destination="./downloads")
    print(f"Downloaded CSV to: {path}")

asyncio.run(main())

Download a grid session as a CSV with a custom file name asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import Grid

syn = Synapse()
syn.login()

async def main():
    grid = Grid(session_id="abc-123-def")
    path = await grid.download_csv_async(
        destination="./downloads", file_name="my_export.csv"
    )
    print(f"Downloaded CSV to: {path}")

asyncio.run(main())

Source code in synapseclient/models/curation.py

@otel_trace_method(
    method_to_trace_name=lambda self, *args, **kwargs: f"Grid_DownloadCsv: ID: {self.session_id}"
)
async def download_csv_async(
    self,
    *,
    destination: Optional[str] = None,
    write_header: bool = True,
    include_row_id_and_row_version: bool = False,
    include_etag: bool = False,
    csv_table_descriptor: Optional[CsvTableDescriptor] = None,
    file_name: Optional[str] = None,
    timeout: int = 120,
    synapse_client: Optional[Synapse] = None,
) -> str:
    """
    Asynchronously download the current state of this grid session as a CSV file.

    Submits a DownloadFromGridRequest async job, waits for it to complete,
    then downloads the resulting CSV to the local filesystem.

    Arguments:
        destination: Local directory path where the CSV will be saved. The directory must already exist.
            If not provided, defaults to the current working directory.
        write_header: Whether the first line should contain column names
            as a header. Defaults to True.
        include_row_id_and_row_version: Whether the first two columns
            should contain row ID and version. Defaults to False.
        include_etag: Whether a column should contain the row etag.
            Defaults to False.
        csv_table_descriptor: The description of the CSV format (delimiter,
            quote character, etc.). If not provided, the default CSV format
            will be used.
        file_name: The optional name for the downloaded file. If not
            provided, defaults to `grid_{session_id}-{timestamp}.csv`.
        timeout: The number of seconds to wait for the async job to
            complete or progress before raising a SynapseTimeoutError.
            Defaults to 120.
        synapse_client: If not passed 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 local path to the downloaded CSV file.

    Raises:
        ValueError: If session_id is not provided.

    Example: Download a grid session as a CSV asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import Grid

        syn = Synapse()
        syn.login()

        async def main():
            grid = Grid(session_id="abc-123-def")
            path = await grid.download_csv_async(destination="./downloads")
            print(f"Downloaded CSV to: {path}")

        asyncio.run(main())
        ```

    Example: Download a grid session as a CSV with a custom file name asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import Grid

        syn = Synapse()
        syn.login()

        async def main():
            grid = Grid(session_id="abc-123-def")
            path = await grid.download_csv_async(
                destination="./downloads", file_name="my_export.csv"
            )
            print(f"Downloaded CSV to: {path}")

        asyncio.run(main())
        ```
    """
    if not self.session_id:
        raise ValueError("session_id is required to download a GridSession as CSV")

    if not destination:
        destination = os.getcwd()

    if not os.path.isdir(destination):
        raise ValueError(f"Destination {destination} is not a valid directory.")

    trace.get_current_span().set_attributes({"synapse.session_id": self.session_id})

    effective_descriptor = csv_table_descriptor or CsvTableDescriptor()
    request = DownloadFromGridRequest(
        session_id=self.session_id,
        write_header=write_header,
        include_row_id_and_row_version=include_row_id_and_row_version,
        include_etag=include_etag,
        csv_table_descriptor=effective_descriptor,
        file_name=file_name,
    )
    download_response = await request.send_job_and_wait_async(
        timeout=timeout, synapse_client=synapse_client
    )
    if not download_response.results_file_handle_id:
        raise ValueError(
            f"Download job for grid session '{self.session_id}' completed but "
            "did not return a file handle ID. The CSV result may be empty or "
            "the job may have failed silently."
        )
    file_handle, presigned_url = await asyncio.gather(
        get_file_handle(
            file_handle_id=download_response.results_file_handle_id,
            synapse_client=synapse_client,
        ),
        get_file_handle_presigned_url(
            file_handle_id=download_response.results_file_handle_id,
            synapse_client=synapse_client,
        ),
    )
    if not file_name:
        timestamp = datetime.now(tz=timezone.utc).strftime("%Y%m%d%H%M%S")
        file_name = f"grid_{self.session_id}-{timestamp}.csv"
    file_path = os.path.join(destination, file_name)
    return await asyncio.to_thread(
        download_from_url,
        url=presigned_url,
        destination=file_path,
        file_handle_id=file_handle["id"],
        expected_md5=file_handle.get("contentMd5"),
        url_is_presigned=True,
        synapse_client=synapse_client,
    )

import_csv_async async

import_csv_async(path: str, *, timeout: int = 120, csv_table_descriptor: Optional[CsvTableDescriptor] = None, synapse_client: Optional[Synapse] = None) -> Grid

Import a CSV file into this grid session. Previews the file to determine the column schema, then imports the data. Currently supports only grids created from a record set.

PARAMETER	DESCRIPTION
path	Local path to the CSV file to import. TYPE: str
csv_table_descriptor	The description of the CSV format (delimiter, quote character, etc.). If not provided, the default CSV format will be used. TYPE: Optional[CsvTableDescriptor] DEFAULT: None
timeout	The number of seconds to wait for each async job to complete or progress before raising a SynapseTimeoutError. Defaults to 120. TYPE: int DEFAULT: 120
synapse_client	If not passed 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
Grid	The Grid object.

RAISES	DESCRIPTION
ValueError	If session_id is not provided.

Import a CSV file into a grid session asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import Grid

syn = Synapse()
syn.login()

async def main():
    grid = Grid(session_id="abc-123-def")
    grid = await grid.import_csv_async(path="/local/path/to/data.csv")
    print(f"Import complete for session: {grid.session_id}")

asyncio.run(main())

Source code in synapseclient/models/curation.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Grid_ImportCsv: ID: {self.session_id}"
)
async def import_csv_async(
    self,
    path: str,
    *,
    timeout: int = 120,
    csv_table_descriptor: Optional[CsvTableDescriptor] = None,
    synapse_client: Optional[Synapse] = None,
) -> "Grid":
    """
    Import a CSV file into this grid session. Previews the file to determine
    the column schema, then imports the data. Currently supports only grids
    created from a record set.

    Arguments:
        path: Local path to the CSV file to import.
        csv_table_descriptor: The description of the CSV format (delimiter,
            quote character, etc.). If not provided, the default CSV format
            will be used.
        timeout: The number of seconds to wait for each async job to complete
            or progress before raising a SynapseTimeoutError. Defaults to 120.
        synapse_client: If not passed 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 Grid object.

    Raises:
        ValueError: If session_id is not provided.

    Example: Import a CSV file into a grid session asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import Grid

        syn = Synapse()
        syn.login()

        async def main():
            grid = Grid(session_id="abc-123-def")
            grid = await grid.import_csv_async(path="/local/path/to/data.csv")
            print(f"Import complete for session: {grid.session_id}")

        asyncio.run(main())
        ```
    """

    if not self.session_id:
        raise ValueError(
            "session_id is required to import a CSV into a GridSession"
        )

    if not os.path.isfile(path):
        raise ValueError(f"Path '{path}' is not a valid file.")

    trace.get_current_span().set_attributes(
        {
            "synapse.session_id": self.session_id,
        }
    )

    client = Synapse.get_client(synapse_client=synapse_client)
    file_handle = await upload_synapse_s3(syn=client, file_path=path)
    file_handle_id = file_handle["id"]

    effective_descriptor = csv_table_descriptor or CsvTableDescriptor()

    upload_to_table_preview = UploadToTablePreviewRequest(
        csv_table_descriptor=effective_descriptor,
        upload_file_handle_id=file_handle_id,
    )

    preview_response = await upload_to_table_preview.send_job_and_wait_async(
        timeout=timeout, synapse_client=synapse_client
    )
    if not preview_response.suggested_columns:
        raise ValueError(
            f"CSV preview for file handle {file_handle_id} returned no suggested "
            f"columns (rows scanned: {preview_response.rows_scanned}). The file may "
            f"be empty, contain only a header row, or use a separator different "
            f"from the configured csv_table_descriptor "
            f"(separator={repr(effective_descriptor.separator)})."
        )

    import_request = GridCsvImportRequest(
        session_id=self.session_id,
        file_handle_id=file_handle_id,
        schema=preview_response.suggested_columns,
        csv_descriptor=effective_descriptor,
    )
    import_response = await import_request.send_job_and_wait_async(
        timeout=timeout, synapse_client=synapse_client
    )
    client.logger.info(
        f"CSV import to grid session {self.session_id} completed successfully, "
        f"total count: {import_response.total_count}, "
        f"total created: {import_response.created_count}, "
        f"total updated: {import_response.updated_count}"
    )

    return self

delete_async async

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

Delete the grid session.

Note: Only the user that created a grid session may delete it.

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 session_id is not provided.

Delete a grid session asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import Grid

syn = Synapse()
syn.login()

async def main():
    # Delete the grid session
    grid = Grid(session_id="abc-123-def")
    await grid.delete_async()
    print("Grid session deleted successfully")

asyncio.run(main())

Source code in synapseclient/models/curation.py

@otel_trace_method(
    method_to_trace_name=lambda self, **kwargs: f"Grid_Delete: ID: {self.session_id}"
)
async def delete_async(self, *, synapse_client: Optional[Synapse] = None) -> None:
    """
    Delete the grid session.

    Note: Only the user that created a grid session may delete it.

    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 session_id is not provided.

    Example: Delete a grid session asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import Grid

        syn = Synapse()
        syn.login()

        async def main():
            # Delete the grid session
            grid = Grid(session_id="abc-123-def")
            await grid.delete_async()
            print("Grid session deleted successfully")

        asyncio.run(main())
        ```
    """
    if not self.session_id:
        raise ValueError("session_id is required to delete a GridSession")

    trace.get_current_span().set_attributes(
        {
            "synapse.session_id": self.session_id or "",
        }
    )

    await delete_grid_session(
        session_id=self.session_id, synapse_client=synapse_client
    )

list_async async classmethod

list_async(source_id: Optional[str] = None, *, synapse_client: Optional[Synapse] = None) -> AsyncGenerator[Grid, None]

Generator to get a list of active grid sessions for the user.

PARAMETER	DESCRIPTION
source_id	Optional. When provided, only sessions with this synId will be returned. 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

YIELDS	DESCRIPTION
AsyncGenerator[Grid, None]	Grid objects representing active grid sessions.

List all active grid sessions asynchronously

 

import asyncio
from synapseclient import Synapse
from synapseclient.models import Grid

syn = Synapse()
syn.login()

async def main():
    # List all active grid sessions for the user
    async for grid in Grid.list_async():
        print(f"Session ID: {grid.session_id}")
        print(f"Source Entity: {grid.source_entity_id}")
        print(f"Started: {grid.started_on}")
        print("---")

    # List grid sessions for a specific source
    async for grid in Grid.list_async(source_id="syn1234567"):
        print(f"Session ID: {grid.session_id}")
        print(f"Modified: {grid.modified_on}")

asyncio.run(main())

Source code in synapseclient/models/curation.py

@skip_async_to_sync
@classmethod
async def list_async(
    cls,
    source_id: Optional[str] = None,
    *,
    synapse_client: Optional[Synapse] = None,
) -> AsyncGenerator["Grid", None]:
    """
    Generator to get a list of active grid sessions for the user.

    Arguments:
        source_id: Optional. When provided, only sessions with this synId will be returned.
        synapse_client: If not passed in and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

    Yields:
        Grid objects representing active grid sessions.

    Example: List all active grid sessions asynchronously
         

        ```python
        import asyncio
        from synapseclient import Synapse
        from synapseclient.models import Grid

        syn = Synapse()
        syn.login()

        async def main():
            # List all active grid sessions for the user
            async for grid in Grid.list_async():
                print(f"Session ID: {grid.session_id}")
                print(f"Source Entity: {grid.source_entity_id}")
                print(f"Started: {grid.started_on}")
                print("---")

            # List grid sessions for a specific source
            async for grid in Grid.list_async(source_id="syn1234567"):
                print(f"Session ID: {grid.session_id}")
                print(f"Modified: {grid.modified_on}")

        asyncio.run(main())
        ```
    """
    async for session_dict in list_grid_sessions(
        source_id=source_id, synapse_client=synapse_client
    ):
        # Convert the dictionary to a Grid object
        grid = cls()
        grid.fill_from_dict(session_dict)
        yield grid

---

synapseclient.models.Query dataclass

Represents a SQL query with optional parameters.

This result is modeled from: https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/table/Query.html

Source code in synapseclient/models/table_components.py

@dataclass
class Query:
    """
    Represents a SQL query with optional parameters.

    This result is modeled from: <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/table/Query.html>
    """

    sql: str
    """The SQL query string"""

    additional_filters: Optional[List[Dict[str, Any]]] = None
    """Appends additional filters to the SQL query. These are applied before facets.
    Filters within the list have an AND relationship. If a WHERE clause already exists
    on the SQL query or facets are selected, it will also be ANDed with the query
    generated by these additional filters."""
    """TODO: create QueryFilter dataclass: https://sagebionetworks.jira.com/browse/SYNPY-1651"""

    selected_facets: Optional[List[Dict[str, Any]]] = None
    """The selected facet filters"""
    """TODO: create FacetColumnRequest dataclass: https://sagebionetworks.jira.com/browse/SYNPY-1651"""

    include_entity_etag: Optional[bool] = False
    """Optional, default false. When true, a query results against views will include
    the Etag of each entity in the results. Note: The etag is necessary to update
    Entities in the view."""

    select_file_column: Optional[int] = None
    """The id of the column used to select file entities (e.g. to fetch the action
    required for download). The column needs to be an ENTITYID type column and be
    part of the schema of the underlying table/view."""

    select_file_version_column: Optional[int] = None
    """The id of the column used as the version for selecting file entities when required
    (e.g. to add a materialized view query to the download cart with version enabled).
    The column needs to be an INTEGER type column and be part of the schema of the
    underlying table/view."""

    offset: Optional[int] = None
    """The optional offset into the results"""

    limit: Optional[int] = None
    """The optional limit to the results"""

    sort: Optional[List[Dict[str, Any]]] = None
    """The sort order for the query results (ARRAY<SortItem>)"""
    """TODO: Add SortItem dataclass: https://sagebionetworks.jira.com/browse/SYNPY-1651 """

    def to_synapse_request(self) -> Dict[str, Any]:
        """Converts the Query object into a dictionary that can be passed into the REST API."""
        result = {
            "sql": self.sql,
            "additionalFilters": self.additional_filters,
            "selectedFacets": self.selected_facets,
            "includeEntityEtag": self.include_entity_etag,
            "selectFileColumn": self.select_file_column,
            "selectFileVersionColumn": self.select_file_version_column,
            "offset": self.offset,
            "limit": self.limit,
            "sort": self.sort,
        }
        delete_none_keys(result)
        return result

Attributes

sql instance-attribute

sql: str

The SQL query string

additional_filters class-attribute instance-attribute

additional_filters: Optional[List[Dict[str, Any]]] = None

Appends additional filters to the SQL query. These are applied before facets. Filters within the list have an AND relationship. If a WHERE clause already exists on the SQL query or facets are selected, it will also be ANDed with the query generated by these additional filters.

selected_facets class-attribute instance-attribute

selected_facets: Optional[List[Dict[str, Any]]] = None

The selected facet filters

include_entity_etag class-attribute instance-attribute

include_entity_etag: Optional[bool] = False

Optional, default false. When true, a query results against views will include the Etag of each entity in the results. Note: The etag is necessary to update Entities in the view.

select_file_column class-attribute instance-attribute

select_file_column: Optional[int] = None

The id of the column used to select file entities (e.g. to fetch the action required for download). The column needs to be an ENTITYID type column and be part of the schema of the underlying table/view.

select_file_version_column class-attribute instance-attribute

select_file_version_column: Optional[int] = None

The id of the column used as the version for selecting file entities when required (e.g. to add a materialized view query to the download cart with version enabled). The column needs to be an INTEGER type column and be part of the schema of the underlying table/view.

offset class-attribute instance-attribute

offset: Optional[int] = None

The optional offset into the results

limit class-attribute instance-attribute

limit: Optional[int] = None

The optional limit to the results

sort class-attribute instance-attribute

sort: Optional[List[Dict[str, Any]]] = None

The sort order for the query results (ARRAY)

---