Table

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.Table dataclass

Bases: TableSynchronousProtocol, AccessControllable

A Table represents the metadata of a table.

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

Source code in synapseclient/models/table.py

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

    Attributes:
        id: The unique immutable ID for this table. A new ID will be generated for new
            Tables. Once issued, this ID is guaranteed to never change or be re-issued
        name: The name of this table. Must be 256 characters or less. Names may only
            contain: letters, numbers, spaces, underscores, hyphens, periods, plus
            signs, apostrophes, and parentheses
        parent_id: The ID of the Entity that is the parent of this table.
        columns: The columns of this table.
        description: The description of this entity. Must be 1000 characters or less.
        etag: Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle
            concurrent updates. Since the E-Tag changes every time an entity is updated
            it is used to detect when a client's current representation of an entity is
            out-of-date.
        created_on: The date this table was created.
        created_by: The ID of the user that created this table.
        modified_on: The date this table was last modified.
            In YYYY-MM-DD-Thh:mm:ss.sssZ format
        modified_by: The ID of the user that last modified this table.
        version_number: The version number issued to this version on the object.
        version_label: The version label for this table
        version_comment: The version comment for this table
        is_latest_version: If this is the latest version of the object.
        is_search_enabled: When creating or updating a table or view specifies if full
            text search should be enabled. Note that enabling full text search might
            slow down the indexing of the table or view.
        activity: The Activity model represents the main record of Provenance in
            Synapse. It is analygous to the Activity defined in the
            [W3C Specification](https://www.w3.org/TR/prov-n/) on Provenance. Activity
            cannot be removed during a store operation by setting it to None. You must
            use: [synapseclient.models.Activity.delete_async][] or
            [synapseclient.models.Activity.disassociate_from_entity_async][].
        annotations: Additional metadata associated with the table. The key is the name
            of your desired annotations. The value is an object containing a list of
            values (use empty list to represent no values for key) and the value type
            associated with all values in the list. To remove all annotations set this
            to an empty dict `{}` or None and store the entity.

    """

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

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

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

    columns: Optional[List[Column]] = None

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

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

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

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

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

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

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

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

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

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

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

    activity: Optional[Activity] = None
    """The Activity model represents the main record of Provenance in Synapse.  It is
    analygous to the Activity defined in the
    [W3C Specification](https://www.w3.org/TR/prov-n/) on Provenance. Activity cannot
    be removed during a store operation by setting it to None. You must use:
    [synapseclient.models.Activity.delete_async][] or
    [synapseclient.models.Activity.disassociate_from_entity_async][].
    """

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

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

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

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

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

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

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

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

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

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

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

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

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

        synapse_schema = Synapse_Schema(
            name=self.name,
            columns=self.columns,
            parent=self.parent_id,
        )
        trace.get_current_span().set_attributes(
            {
                "synapse.name": self.name or "",
                "synapse.id": self.id or "",
            }
        )
        loop = asyncio.get_event_loop()
        entity = await loop.run_in_executor(
            None,
            lambda: Synapse.get_client(synapse_client=synapse_client).store(
                obj=synapse_schema
            ),
        )

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

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

        return self

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

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

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

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

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

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

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

        Arguments:
            query: The query to run.
            result_format: The format of the results. Defaults to CsvResultFormat().
            synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

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

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

Functions

get_async async

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

Get the metadata about the table from synapse.

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

RETURNS	DESCRIPTION
Table	The Table instance stored in synapse.

Source code in synapseclient/models/table.py

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

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

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

store_schema_async async

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

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

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

RETURNS	DESCRIPTION
Table	The Table instance stored in synapse.

Source code in synapseclient/models/table.py

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

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

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

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

    synapse_schema = Synapse_Schema(
        name=self.name,
        columns=self.columns,
        parent=self.parent_id,
    )
    trace.get_current_span().set_attributes(
        {
            "synapse.name": self.name or "",
            "synapse.id": self.id or "",
        }
    )
    loop = asyncio.get_event_loop()
    entity = await loop.run_in_executor(
        None,
        lambda: Synapse.get_client(synapse_client=synapse_client).store(
            obj=synapse_schema
        ),
    )

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

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

    return self

store_rows_from_csv_async async

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

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

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

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

Source code in synapseclient/models/table.py

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

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

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

delete_rows_async async

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

Delete rows from a table.

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

RETURNS	DESCRIPTION
None	None

Source code in synapseclient/models/table.py

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

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

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

query_async async classmethod

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

Query for data on a table stored in Synapse.

PARAMETER	DESCRIPTION
query	The query to run. TYPE: str
result_format	The format of the results. Defaults to CsvResultFormat(). TYPE: Union[CsvResultFormat, RowsetResultFormat] DEFAULT: CsvResultFormat()
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

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

Source code in synapseclient/models/table.py

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

    Arguments:
        query: The query to run.
        result_format: The format of the results. Defaults to CsvResultFormat().
        synapse_client: If not passed in and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

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

delete_async async

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

Delete the table from synapse.

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

RETURNS	DESCRIPTION
None	None

Source code in synapseclient/models/table.py

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

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

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

get_permissions_async async

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

Get the permissions that the caller has on an Entity.

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

RETURNS	DESCRIPTION
Permissions	A Permissions object

Using this function:

Getting permissions for a Synapse Entity

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

Getting access types list from the Permissions object

permissions.access_types

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

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

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

    Returns:
        A Permissions object


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

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

        Getting access types list from the Permissions object

            permissions.access_types
    """
    loop = asyncio.get_event_loop()

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

get_acl_async async

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

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

PARAMETER	DESCRIPTION
principal_id	Identifier of a user or group (defaults to PUBLIC users) TYPE: int DEFAULT: None
synapse_client	If not passed in and caching was not disabled by Synapse.allow_client_caching(False) this will use the last created instance from the Synapse class constructor. TYPE: Optional[Synapse] DEFAULT: None

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

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

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

    Arguments:
        principal_id: Identifier of a user or group (defaults to PUBLIC users)
        synapse_client: If not passed in and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

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

set_permissions_async async

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

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

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

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

Setting permissions

Grant all registered users download access

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

Grant the public view access

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

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

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

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

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

    Returns:
        An Access Control List object

    Example: Setting permissions
        Grant all registered users download access

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

        Grant the public view access

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

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