Activity

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.

Example Script

Working with activities

"""The purpose of this script is to demonstrate how to use the OOP interface for Activity.
The following actions are shown in this script:
1. Creating a file with an Activity
2. Retrieve an activity by parent File
3. Creating a second file with the same activity
4. Modifying the activity attached to both Files
5. Creating a table with an Activity
"""

import os

import synapseclient
from synapseclient.models import (
    Activity,
    Column,
    ColumnType,
    File,
    Table,
    UsedEntity,
    UsedURL,
)

PROJECT_ID = "syn52948289"

syn = synapseclient.Synapse(debug=True)
syn.login()


def create_random_file(
    path: str,
) -> None:
    """Create a random file with random data.

    Arguments:
        path: The path to create the file at.
    """
    with open(path, "wb") as f:
        f.write(os.urandom(1))


def store_activity_on_file():
    name_of_file = "my_file_with_random_data.txt"
    path_to_file = os.path.join(os.path.expanduser("~/temp"), name_of_file)
    create_random_file(path_to_file)

    name_of_second_file = "my_second_file_with_random_data.txt"
    path_to_second_file = os.path.join(
        os.path.expanduser("~/temp"), name_of_second_file
    )
    create_random_file(path_to_second_file)

    # Create an activity =================================================================
    activity = Activity(
        name="My Activity",
        description="This is an activity.",
        used=[
            UsedURL(name="Used URL", url="https://www.synapse.org/"),
            UsedEntity(target_id=PROJECT_ID),
        ],
        executed=[
            UsedURL(name="Used URL", url="https://www.synapse.org/"),
            UsedEntity(target_id=PROJECT_ID),
        ],
    )

    # Creating and uploading a file to a project =========================================
    file = File(
        path=path_to_file,
        name=name_of_file,
        parent_id=PROJECT_ID,
        description="This is a file with random data.",
        activity=activity,
    )

    file = file.store()

    print("File created with activity:")
    print(activity)

    activity_copy = Activity.from_parent(parent=file)

    # Storing a second file to a project and re-use the activity =========================
    second_file = File(
        path=path_to_second_file,
        name=name_of_second_file,
        parent_id=PROJECT_ID,
        description="This is a file with random data.",
        activity=activity_copy,
    )

    second_file.store()

    print("Second file created with activity:")
    print(second_file.activity)

    # # Update the already created activity, which updates the activity on both files ====
    new_activity_instance = Activity(
        # In order to update an existing activity you must provide the id and etag.
        id=second_file.activity.id,
        etag=second_file.activity.etag if second_file.activity else None,
        name="My Activity - MODIFIED",
        used=[
            UsedURL(name="Used URL", url="https://www.synapse.org/"),
            UsedEntity(target_id=PROJECT_ID),
        ],
        executed=[
            UsedURL(name="Used URL", url="https://www.synapse.org/"),
            UsedEntity(target_id=PROJECT_ID),
        ],
    )
    # These 3 will be the same activity
    print("Modified activity showing name is updated for all files:")
    print(new_activity_instance.store().name)
    print(Activity.from_parent(parent=file).name)
    print(Activity.from_parent(parent=second_file).name)


def store_activity_on_table():
    # Create an activity =================================================================
    activity = Activity(
        name="My Activity",
        description="This is an activity.",
        used=[
            UsedURL(name="Used URL", url="https://www.synapse.org/"),
            UsedEntity(target_id=PROJECT_ID),
        ],
        executed=[
            UsedURL(name="Used URL", url="https://www.synapse.org/"),
            UsedEntity(target_id=PROJECT_ID),
        ],
    )

    # Creating columns for my table ======================================================
    columns = [
        Column(id=None, name="my_string_column", column_type=ColumnType.STRING),
    ]

    # Creating a table ===============================================================
    table = Table(
        name="my_first_test_table",
        columns=columns,
        parent_id=PROJECT_ID,
        activity=activity,
    )

    table = table.store_schema()

    print("Table created with activity:")
    print(table.activity)


store_activity_on_file()
store_activity_on_table()

API Reference

synapseclient.models.Activity dataclass

Bases: ActivitySynchronousProtocol

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

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

Source code in synapseclient/models/activity.py

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

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

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

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

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

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

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

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

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

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

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

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

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

        Arguments:
            synapse_activity: The response from the REST API.

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

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

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

        return self

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

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

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

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

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

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

        Returns:
            The activity object.

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

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

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

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

        return self

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

        Arguments:
            parent: The parent entity this activity is associated with. The parent may
                also have a version_number. Gets the most recent version if version is
                omitted.
            synapse_client: If not passed in and caching was not disabled by
                `Synapse.allow_client_caching(False)` this will use the last created
                instance from the Synapse class constructor.

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

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

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

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

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

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

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

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

Functions

from_parent classmethod

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

Get the Activity from Synapse based on the parent entity.

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

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

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

Source code in synapseclient/models/protocols/activity_protocol.py

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

    Arguments:
        parent: The parent entity this activity is associated with. The parent may
            also have a version_number. Gets the most recent version if version is
            omitted.
        synapse_client: If not passed in and caching was not disabled by
            `Synapse.allow_client_caching(False)` this will use the last created
            instance from the Synapse class constructor.

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

    Raises:
        ValueError: If the parent does not have an ID.
    """
    from synapseclient.models import Activity

    return Activity()

store

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

Store the Activity in Synapse.

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

RETURNS	DESCRIPTION
Activity	The activity object.

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

Source code in synapseclient/models/protocols/activity_protocol.py

def store(
    self,
    parent: Optional[Union["Table", "File"]] = None,
    *,
    synapse_client: Optional[Synapse] = None,
) -> "Activity":
    """
    Store the Activity in Synapse.

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

    Returns:
        The activity object.

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

            - If the parent does not have an ID.
            - If the Activity does not have an ID and ETag.
    """
    return self

delete classmethod

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

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

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

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

Source code in synapseclient/models/protocols/activity_protocol.py

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

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

    Raises:
        ValueError: If the parent does not have an ID.
    """
    return None

---

synapseclient.models.UsedEntity dataclass

Reference to a Synapse entity that was used or executed by an Activity.

ATTRIBUTE	DESCRIPTION
target_id	The ID of the entity to which this reference refers. TYPE: Optional[str]
target_version_number	The version number of the entity to which this reference refers. TYPE: Optional[int]

Source code in synapseclient/models/activity.py

@dataclass
class UsedEntity:
    """
    Reference to a Synapse entity that was used or executed by an Activity.

    Attributes:
        target_id: The ID of the entity to which this reference refers.
        target_version_number: The version number of the entity to which this reference refers.
    """

    target_id: Optional[str] = None
    """The the id of the entity to which this reference refers."""

    target_version_number: Optional[int] = None
    """The version number of the entity to which this reference refers."""

    def format_for_manifest(self) -> str:
        """
        Format the content of this data class to be written to a manifest file.

        Returns:
            The formatted string.
        """
        return_value = f"{self.target_id}"
        if self.target_version_number is not None:
            return_value += f".{self.target_version_number}"
        return return_value

---

synapseclient.models.UsedURL dataclass

URL that was used or executed by an Activity such as a link to a GitHub commit or a link to a specific version of a software tool.

ATTRIBUTE	DESCRIPTION
name	The name of the URL. TYPE: Optional[str]
url	The external URL of the file that was used such as a link to a GitHub commit or a link to a specific version of a software tool. TYPE: Optional[str]

Source code in synapseclient/models/activity.py

@dataclass
class UsedURL:
    """
    URL that was used or executed by an Activity such as a link to a
    GitHub commit or a link to a specific version of a software tool.

    Attributes:
        name: The name of the URL.
        url: The external URL of the file that was used such as a link to a
            GitHub commit or a link to a specific version of a software tool.
    """

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

    url: Optional[str] = None
    """The external URL of the file that was used such as a link to a GitHub commit
    or a link to a specific version of a software tool."""

    def format_for_manifest(self) -> str:
        """
        Format the content of this data class to be written to a manifest file.

        Returns:
            The formatted string.
        """
        if self.name:
            return_value = self.name
        else:
            return_value = self.url

        return return_value