schematic.store.synapse

   1"""Synapse storage class"""
   2
   3import asyncio
   4import atexit
   5import logging
   6import os
   7import re
   8import secrets
   9import shutil
  10import time
  11import uuid  # used to generate unique names for entities
  12from copy import deepcopy
  13from dataclasses import dataclass, field
  14from time import sleep
  15
  16# allows specifying explicit variable types
  17from typing import Any, Dict, List, Optional, Sequence, Set, Tuple, Union
  18
  19import numpy as np
  20import pandas as pd
  21import synapseclient
  22from opentelemetry import trace
  23from synapseclient import Annotations as OldAnnotations
  24from synapseclient import (
  25    Column,
  26    EntityViewSchema,
  27    EntityViewType,
  28    File,
  29    Folder,
  30    Schema,
  31    Synapse,
  32    Table,
  33    as_table_columns,
  34)
  35from synapseclient.annotations import _convert_to_annotations_list
  36from synapseclient.api import get_config_file, get_entity_id_bundle2
  37from synapseclient.core.constants.concrete_types import PROJECT_ENTITY
  38from synapseclient.core.exceptions import (
  39    SynapseAuthenticationError,
  40    SynapseHTTPError,
  41    SynapseUnmetAccessRestrictions,
  42)
  43from synapseclient.models.annotations import Annotations
  44from synapseclient.table import CsvFileTable, Schema, build_table
  45from tenacity import (
  46    retry,
  47    retry_if_exception_type,
  48    stop_after_attempt,
  49    wait_chain,
  50    wait_fixed,
  51)
  52
  53from schematic.configuration.configuration import CONFIG
  54from schematic.exceptions import AccessCredentialsError
  55from schematic.schemas.data_model_graph import DataModelGraphExplorer
  56from schematic.store.base import BaseStorage
  57from schematic.store.database.synapse_database import SynapseDatabase
  58from schematic.store.synapse_tracker import SynapseEntityTracker
  59from schematic.utils.df_utils import (
  60    STR_NA_VALUES_FILTERED,
  61    col_in_dataframe,
  62    load_df,
  63    update_df,
  64)
  65
  66# entity_type_mapping, get_dir_size, create_temp_folder, check_synapse_cache_size, and clear_synapse_cache functions are used for AWS deployment
  67# Please do not remove these import statements
  68from schematic.utils.general import (
  69    check_synapse_cache_size,
  70    clear_synapse_cache,
  71    create_temp_folder,
  72    entity_type_mapping,
  73    get_dir_size,
  74)
  75from schematic.utils.io_utils import cleanup_temporary_storage
  76from schematic.utils.schema_utils import get_class_label_from_display_name
  77from schematic.utils.validate_utils import comma_separated_list_regex, rule_in_rule_list
  78
  79logger = logging.getLogger("Synapse storage")
  80
  81tracer = trace.get_tracer("Schematic")
  82
  83
  84@dataclass
  85class ManifestDownload(object):
  86    """
  87    syn: an object of type synapseclient.
  88    manifest_id: id of a manifest
  89    synapse_entity_tracker: Tracker for a pull-through cache of Synapse entities
  90    """
  91
  92    syn: synapseclient.Synapse
  93    manifest_id: str
  94    synapse_entity_tracker: SynapseEntityTracker = field(
  95        default_factory=SynapseEntityTracker
  96    )
  97
  98    def _download_manifest_to_folder(self, use_temporary_folder: bool = True) -> File:
  99        """
 100        Try downloading a manifest to a specific folder (temporary or not). When the
 101        `use_temporary_folder` is set to True, the manifest will be downloaded to a
 102        temporary folder. This is useful for when the code is running as an API server
 103        where multiple requests are being made at the same time. This will prevent
 104        multiple requests from overwriting the same manifest file. When the
 105        `use_temporary_folder` is set to False, the manifest will be downloaded to the
 106        default manifest folder.
 107
 108        Args:
 109            use_temporary_folder: boolean argument indicating if a temporary folder
 110                should be used to store the manifest file. This is useful when running
 111                this code as an API server where multiple requests could be made at the
 112                same time. This is set to False when the code is being used from the
 113                CLI. Defaults to True.
 114
 115        Return:
 116            manifest_data: A Synapse file entity of the downloaded manifest
 117        """
 118        manifest_data = self.synapse_entity_tracker.get(
 119            synapse_id=self.manifest_id,
 120            syn=self.syn,
 121            download_file=False,
 122            retrieve_if_not_present=False,
 123        )
 124        current_span = trace.get_current_span()
 125        if (
 126            manifest_data
 127            and (file_handle := manifest_data.get("_file_handle", None))
 128            and current_span.is_recording()
 129        ):
 130            current_span.set_attribute(
 131                "schematic.manifest_size", file_handle.get("contentSize", 0)
 132            )
 133
 134        if manifest_data and manifest_data.path:
 135            return manifest_data
 136
 137        if "SECRETS_MANAGER_SECRETS" in os.environ:
 138            temporary_manifest_storage = "/var/tmp/temp_manifest_download"
 139            cleanup_temporary_storage(
 140                temporary_manifest_storage, time_delta_seconds=3600
 141            )
 142            # create a new directory to store manifest
 143            if not os.path.exists(temporary_manifest_storage):
 144                os.mkdir(temporary_manifest_storage)
 145            # create temporary folders for storing manifests
 146            download_location = create_temp_folder(
 147                path=temporary_manifest_storage,
 148                prefix=f"{self.manifest_id}-{time.time()}-",
 149            )
 150        else:
 151            if use_temporary_folder:
 152                download_location = create_temp_folder(
 153                    path=CONFIG.manifest_folder,
 154                    prefix=f"{self.manifest_id}-{time.time()}-",
 155                )
 156            else:
 157                download_location = CONFIG.manifest_folder
 158
 159        manifest_data = self.synapse_entity_tracker.get(
 160            synapse_id=self.manifest_id,
 161            syn=self.syn,
 162            download_file=True,
 163            retrieve_if_not_present=True,
 164            download_location=download_location,
 165        )
 166
 167        # This is doing a rename of the downloaded file. The reason this is important
 168        # is that if we are re-using a file that was previously downloaded, but the
 169        # file had been renamed. The file downloaded from the Synapse client is just
 170        # a direct copy of that renamed file. This code will set the name of the file
 171        # to the original name that was used to download the file. Note: An MD5 checksum
 172        # of the file will still be performed so if the file has changed, it will be
 173        # downloaded again.
 174        filename = manifest_data._file_handle.fileName
 175        if filename != os.path.basename(manifest_data.path):
 176            parent_folder = os.path.dirname(manifest_data.path)
 177            manifest_original_name_and_path = os.path.join(parent_folder, filename)
 178
 179            self.syn.cache.remove(
 180                file_handle_id=manifest_data.dataFileHandleId, path=manifest_data.path
 181            )
 182            os.rename(manifest_data.path, manifest_original_name_and_path)
 183            manifest_data.path = manifest_original_name_and_path
 184            self.syn.cache.add(
 185                file_handle_id=manifest_data.dataFileHandleId,
 186                path=manifest_original_name_and_path,
 187                md5=manifest_data._file_handle.contentMd5,
 188            )
 189
 190        return manifest_data
 191
 192    def _entity_type_checking(self) -> str:
 193        """
 194        check the entity type of the id that needs to be downloaded
 195        Return:
 196             if the entity type is wrong, raise an error
 197        """
 198        # check the type of entity
 199        entity_type = entity_type_mapping(
 200            syn=self.syn,
 201            entity_id=self.manifest_id,
 202            synapse_entity_tracker=self.synapse_entity_tracker,
 203        )
 204        if entity_type != "file":
 205            logger.error(
 206                f"You are using entity type: {entity_type}. Please provide a file ID"
 207            )
 208
 209    def download_manifest(
 210        self,
 211        newManifestName: str = "",
 212        manifest_df: pd.DataFrame = pd.DataFrame(),
 213        use_temporary_folder: bool = True,
 214    ) -> Union[str, File]:
 215        """
 216        Download a manifest based on a given manifest id.
 217        Args:
 218            newManifestName(optional): new name of a manifest that gets downloaded.
 219            manifest_df(optional): a dataframe containing name and id of manifests in a given asset view
 220        Return:
 221            manifest_data: synapse entity file object
 222        """
 223
 224        # enables retrying if user does not have access to uncensored manifest
 225        # pass synID to synapseclient.Synapse.get() method to download (and overwrite) file to a location
 226        manifest_data = ""
 227
 228        # check entity type
 229        self._entity_type_checking()
 230
 231        # download a manifest
 232        try:
 233            manifest_data = self._download_manifest_to_folder(
 234                use_temporary_folder=use_temporary_folder
 235            )
 236        except (SynapseUnmetAccessRestrictions, SynapseAuthenticationError):
 237            # if there's an error getting an uncensored manifest, try getting the censored manifest
 238            if not manifest_df.empty:
 239                censored_regex = re.compile(".*censored.*")
 240                censored = manifest_df["name"].str.contains(censored_regex)
 241                new_manifest_id = manifest_df[censored]["id"][0]
 242                self.manifest_id = new_manifest_id
 243                try:
 244                    manifest_data = self._download_manifest_to_folder(
 245                        use_temporary_folder=use_temporary_folder
 246                    )
 247                except (
 248                    SynapseUnmetAccessRestrictions,
 249                    SynapseAuthenticationError,
 250                ) as e:
 251                    raise PermissionError(
 252                        "You don't have access to censored and uncensored manifests in this dataset."
 253                    ) from e
 254            else:
 255                logger.error(
 256                    f"You don't have access to the requested resource: {self.manifest_id}"
 257                )
 258
 259        if newManifestName and os.path.exists(manifest_data.get("path")):
 260            # Rename the file we just made to the new name
 261            new_manifest_filename = newManifestName + ".csv"
 262
 263            # get location of existing manifest. The manifest that will be renamed should live in the same folder as existing manifest.
 264            parent_folder = os.path.dirname(manifest_data.get("path"))
 265
 266            new_manifest_path_name = os.path.join(parent_folder, new_manifest_filename)
 267
 268            # Copy file to new location. The purpose of using a copy instead of a rename
 269            # is to avoid any potential issues with the file being used in another
 270            # process. This avoids any potential race or code cocurrency conditions.
 271            shutil.copyfile(src=manifest_data["path"], dst=new_manifest_path_name)
 272
 273            # Adding this to cache will allow us to re-use the already downloaded
 274            # manifest file for up to 1 hour.
 275            self.syn.cache.add(
 276                file_handle_id=manifest_data.dataFileHandleId,
 277                path=new_manifest_path_name,
 278                md5=manifest_data._file_handle.contentMd5,
 279            )
 280
 281            # Update file names/paths in manifest_data
 282            manifest_data["name"] = new_manifest_filename
 283            manifest_data["filename"] = new_manifest_filename
 284            manifest_data["path"] = new_manifest_path_name
 285
 286        return manifest_data
 287
 288
 289class SynapseStorage(BaseStorage):
 290    """Implementation of Storage interface for datasets/files stored on Synapse.
 291    Provides utilities to list files in a specific project; update files annotations, create fileviews, etc.
 292
 293    TODO: Need to define the interface and rename and/or refactor some of the methods below.
 294    """
 295
 296    @tracer.start_as_current_span("SynapseStorage::__init__")
 297    def __init__(
 298        self,
 299        token: Optional[str] = None,  # optional parameter retrieved from browser cookie
 300        access_token: Optional[str] = None,
 301        project_scope: Optional[list] = None,
 302        synapse_cache_path: Optional[str] = None,
 303        perform_query: Optional[bool] = True,
 304        columns: Optional[list] = None,
 305        where_clauses: Optional[list] = None,
 306    ) -> None:
 307        """Initializes a SynapseStorage object.
 308
 309        Args:
 310            token (Optional[str], optional):
 311              Optional token parameter as found in browser cookie upon login to synapse.
 312              Defaults to None.
 313            access_token (Optional[list], optional):
 314              Optional access token (personal or oauth).
 315              Defaults to None.
 316            project_scope (Optional[list], optional): Defaults to None.
 317            synapse_cache_path (Optional[str], optional):
 318              Location of synapse cache.
 319              Defaults to None.
 320        TODO:
 321            Consider necessity of adding "columns" and "where_clauses" params to the constructor. Currently with how `query_fileview` is implemented, these params are not needed at this step but could be useful in the future if the need for more scoped querys expands.
 322        """
 323        self.syn = self.login(synapse_cache_path, access_token)
 324        current_span = trace.get_current_span()
 325        if current_span.is_recording():
 326            current_span.set_attribute("user.id", self.syn.credentials.owner_id)
 327        self.project_scope = project_scope
 328        self.storageFileview = CONFIG.synapse_master_fileview_id
 329        self.manifest = CONFIG.synapse_manifest_basename
 330        self.root_synapse_cache = self.syn.cache.cache_root_dir
 331        self.synapse_entity_tracker = SynapseEntityTracker()
 332        if perform_query:
 333            self.query_fileview(columns=columns, where_clauses=where_clauses)
 334
 335    # TODO: When moving this over to a regular cron-job the following logic should be
 336    # out of `manifest_download`:
 337    # if "SECRETS_MANAGER_SECRETS" in os.environ:
 338    #     temporary_manifest_storage = "/var/tmp/temp_manifest_download"
 339    #     cleanup_temporary_storage(temporary_manifest_storage, time_delta_seconds=3600)
 340    @tracer.start_as_current_span("SynapseStorage::_purge_synapse_cache")
 341    def _purge_synapse_cache(
 342        self, maximum_storage_allowed_cache_gb: int = 1, minute_buffer: int = 15
 343    ) -> None:
 344        """
 345        Purge synapse cache if it exceeds a certain size. Default to 1GB.
 346        Args:
 347            maximum_storage_allowed_cache_gb (int): the maximum storage allowed
 348              before purging cache. Default is 1 GB.
 349            minute_buffer (int): All files created this amount of time or older will be deleted
 350        """
 351        # try clearing the cache
 352        # scan a directory and check size of files
 353        if os.path.exists(self.root_synapse_cache):
 354            maximum_storage_allowed_cache_bytes = maximum_storage_allowed_cache_gb * (
 355                1024**3
 356            )
 357            nbytes = get_dir_size(self.root_synapse_cache)
 358            dir_size_bytes = check_synapse_cache_size(directory=self.root_synapse_cache)
 359            # if 1 GB has already been taken, purge cache before 15 min
 360            if dir_size_bytes >= maximum_storage_allowed_cache_bytes:
 361                num_of_deleted_files = clear_synapse_cache(
 362                    self.syn.cache, minutes=minute_buffer
 363                )
 364                logger.info(
 365                    f"{num_of_deleted_files}  files have been deleted from {self.root_synapse_cache}"
 366                )
 367            else:
 368                # on AWS, OS takes around 14-17% of our ephemeral storage (20GiB)
 369                # instead of guessing how much space that we left, print out .synapseCache here
 370                logger.info(f"the total size of .synapseCache is: {nbytes} bytes")
 371
 372    @tracer.start_as_current_span("SynapseStorage::query_fileview")
 373    def query_fileview(
 374        self,
 375        columns: Optional[list] = None,
 376        where_clauses: Optional[list] = None,
 377        force_requery: Optional[bool] = False,
 378    ) -> None:
 379        """
 380        Method to query the Synapse FileView and store the results in a pandas DataFrame. The results are stored in the storageFileviewTable attribute.
 381        Is called once during initialization of the SynapseStorage object and can be called again later to specify a specific, more limited scope for validation purposes.
 382        Args:
 383            columns (Optional[list], optional): List of columns to be selected from the table. Defaults behavior is to request all columns.
 384            where_clauses (Optional[list], optional): List of where clauses to be used to scope the query. Defaults to None.
 385            force_requery (Optional[bool], optional): If True, forces a requery of the fileview. Defaults to False.
 386        """
 387        self._purge_synapse_cache()
 388
 389        # Initialize to assume that the new fileview query will be different from what may already be stored. Initializes to True because generally one will not have already been performed
 390        self.new_query_different = True
 391
 392        # If a query has already been performed, store the query
 393        previous_query_built = hasattr(self, "fileview_query")
 394        if previous_query_built:
 395            previous_query = self.fileview_query
 396
 397        # Build a query with the current given parameters and check to see if it is different from the previous
 398        self._build_query(columns=columns, where_clauses=where_clauses)
 399        if previous_query_built:
 400            self.new_query_different = self.fileview_query != previous_query
 401
 402        # Only perform the query if it is different from the previous query or we are forcing new results to be retrieved
 403        if self.new_query_different or force_requery:
 404            try:
 405                self.storageFileviewTable = self.syn.tableQuery(
 406                    query=self.fileview_query,
 407                ).asDataFrame(na_values=STR_NA_VALUES_FILTERED, keep_default_na=False)
 408            except SynapseHTTPError as exc:
 409                exception_text = str(exc)
 410                if "Unknown column path" in exception_text:
 411                    raise ValueError(
 412                        "The path column has not been added to the fileview. Please make sure that the fileview is up to date. You can add the path column to the fileview by follwing the instructions in the validation rules documentation."
 413                    )
 414                elif "Unknown column" in exception_text:
 415                    missing_column = exception_text.split("Unknown column ")[-1]
 416                    raise ValueError(
 417                        f"The columns {missing_column} specified in the query do not exist in the fileview. Please make sure that the column names are correct and that all expected columns have been added to the fileview."
 418                    )
 419                else:
 420                    raise AccessCredentialsError(self.storageFileview)
 421
 422    @staticmethod
 423    def build_clause_from_dataset_id(
 424        dataset_id: Optional[str] = None, dataset_folder_list: Optional[list] = None
 425    ) -> str:
 426        """
 427        Method to build a where clause for a Synapse FileView query based on a dataset ID that can be used before an object is initialized.
 428        Args:
 429            dataset_id: Synapse ID of a dataset that should be used to limit the query
 430            dataset_folder_list: List of Synapse IDs of a dataset and all its subfolders that should be used to limit the query
 431        Returns:
 432            clause for the query or an empty string if no dataset ID is provided
 433        """
 434        # Calling this method without specifying synIDs will complete but will not scope the view
 435        if (not dataset_id) and (not dataset_folder_list):
 436            return ""
 437
 438        # This will be used to gather files under a dataset recursively with a fileview query instead of walking
 439        if dataset_folder_list:
 440            search_folders = ", ".join(f"'{synId}'" for synId in dataset_folder_list)
 441            return f"parentId IN ({search_folders})"
 442
 443        # `dataset_id` should be provided when all files are stored directly under the dataset folder
 444        return f"parentId='{dataset_id}'"
 445
 446    def _build_query(
 447        self, columns: Optional[list] = None, where_clauses: Optional[list] = None
 448    ):
 449        """
 450        Method to build a query for Synapse FileViews
 451        Args:
 452            columns (Optional[list], optional): List of columns to be selected from the table. Defaults behavior is to request all columns.
 453            where_clauses (Optional[list], optional): List of where clauses to be used to scope the query. Defaults to None.
 454            self.storageFileview (str): Synapse FileView ID
 455            self.project_scope (Optional[list], optional): List of project IDs to be used to scope the query. Defaults to None.
 456                Gets added to where_clauses, more included for backwards compatability and as a more user friendly way of subsetting the view in a simple way.
 457        """
 458        if columns is None:
 459            columns = []
 460        if where_clauses is None:
 461            where_clauses = []
 462
 463        if self.project_scope:
 464            project_scope_clause = f"projectId IN {tuple(self.project_scope + [''])}"
 465            where_clauses.append(project_scope_clause)
 466
 467        if where_clauses:
 468            where_clauses = " AND ".join(where_clauses)
 469            where_clauses = f"WHERE {where_clauses} ;"
 470        else:
 471            where_clauses = ";"
 472
 473        if columns:
 474            columns = ",".join(columns)
 475        else:
 476            columns = "*"
 477
 478        self.fileview_query = (
 479            f"SELECT {columns} FROM {self.storageFileview} {where_clauses}"
 480        )
 481
 482        return
 483
 484    @staticmethod
 485    @tracer.start_as_current_span("SynapseStorage::login")
 486    def login(
 487        synapse_cache_path: Optional[str] = None,
 488        access_token: Optional[str] = None,
 489    ) -> synapseclient.Synapse:
 490        """Login to Synapse
 491
 492        Args:
 493            access_token (Optional[str], optional): A synapse access token. Defaults to None.
 494            synapse_cache_path (Optional[str]): location of synapse cache
 495
 496        Raises:
 497            ValueError: If unable to loging with access token
 498
 499        Returns:
 500            synapseclient.Synapse: A Synapse object that is logged in
 501        """
 502        # If no token is provided, try retrieving access token from environment
 503        if not access_token:
 504            access_token = os.getenv("SYNAPSE_ACCESS_TOKEN")
 505
 506        # login using a token
 507        if access_token:
 508            try:
 509                syn = synapseclient.Synapse(
 510                    cache_root_dir=synapse_cache_path,
 511                    debug=False,
 512                    skip_checks=True,
 513                    cache_client=False,
 514                )
 515                syn.login(authToken=access_token, silent=True)
 516                current_span = trace.get_current_span()
 517                if current_span.is_recording():
 518                    current_span.set_attribute("user.id", syn.credentials.owner_id)
 519            except SynapseHTTPError as exc:
 520                raise ValueError(
 521                    "No access to resources. Please make sure that your token is correct"
 522                ) from exc
 523        else:
 524            # login using synapse credentials provided by user in .synapseConfig (default) file
 525            syn = synapseclient.Synapse(
 526                configPath=CONFIG.synapse_configuration_path,
 527                cache_root_dir=synapse_cache_path,
 528                debug=False,
 529                skip_checks=True,
 530                cache_client=False,
 531            )
 532            syn.login(silent=True)
 533            current_span = trace.get_current_span()
 534            if current_span.is_recording():
 535                current_span.set_attribute("user.id", syn.credentials.owner_id)
 536        return syn
 537
 538    def missing_entity_handler(method):
 539        def wrapper(*args, **kwargs):
 540            try:
 541                return method(*args, **kwargs)
 542            except SynapseHTTPError as ex:
 543                str_message = str(ex).replace("\n", "")
 544                if "trash" in str_message or "does not exist" in str_message:
 545                    logging.warning(str_message)
 546                    return None
 547                else:
 548                    raise ex
 549
 550        return wrapper
 551
 552    def async_missing_entity_handler(method):
 553        """Decorator to handle missing entities in async methods."""
 554
 555        async def wrapper(*args: Any, **kwargs: Any) -> Any:
 556            try:
 557                return await method(*args, **kwargs)
 558            except SynapseHTTPError as ex:
 559                str_message = str(ex).replace("\n", "")
 560                if "trash" in str_message or "does not exist" in str_message:
 561                    logging.warning(str_message)
 562                    return None
 563                else:
 564                    raise ex
 565
 566        return wrapper
 567
 568    def getStorageFileviewTable(self):
 569        """Returns the storageFileviewTable obtained during initialization."""
 570        return self.storageFileviewTable
 571
 572    def getPaginatedRestResults(self, currentUserId: str) -> Dict[str, str]:
 573        """Gets the paginated results of the REST call to Synapse to check what projects the current user has access to.
 574
 575        Args:
 576            currentUserId: synapse id for the user whose projects we want to get.
 577
 578        Returns:
 579            A dictionary with a next page token and the results.
 580        """
 581        all_results = self.syn.restGET(
 582            "/projects/user/{principalId}".format(principalId=currentUserId)
 583        )
 584
 585        while (
 586            "nextPageToken" in all_results
 587        ):  # iterate over next page token in results while there is any
 588            results_token = self.syn.restGET(
 589                "/projects/user/{principalId}?nextPageToken={nextPageToken}".format(
 590                    principalId=currentUserId,
 591                    nextPageToken=all_results["nextPageToken"],
 592                )
 593            )
 594            all_results["results"].extend(results_token["results"])
 595
 596            if "nextPageToken" in results_token:
 597                all_results["nextPageToken"] = results_token["nextPageToken"]
 598            else:
 599                del all_results["nextPageToken"]
 600
 601        return all_results
 602
 603    @tracer.start_as_current_span("SynapseStorage::getStorageProjects")
 604    def getStorageProjects(self, project_scope: List = None) -> list[tuple[str, str]]:
 605        """Gets all storage projects the current user has access to, within the scope of the 'storageFileview' attribute.
 606
 607        Returns:
 608            A list of storage projects the current user has access to; the list consists of tuples (projectId, projectName).
 609        """
 610
 611        # get the set of all storage Synapse project accessible for this pipeline
 612        storageProjects = self.storageFileviewTable["projectId"].unique()
 613
 614        # get the set of storage Synapse project accessible for this user
 615        # get a list of projects from Synapse
 616        current_user_project_headers = self.synapse_entity_tracker.get_project_headers(
 617            current_user_id=self.syn.credentials.owner_id, syn=self.syn
 618        )
 619        project_id_to_name_dict = {}
 620        current_user_projects = []
 621        for project_header in current_user_project_headers:
 622            project_id_to_name_dict[project_header.get("id")] = project_header.get(
 623                "name"
 624            )
 625            current_user_projects.append(project_header.get("id"))
 626
 627        # find set of user projects that are also in this pipeline's storage projects set
 628        storageProjects = list(set(storageProjects) & set(current_user_projects))
 629
 630        # Limit projects to scope if specified
 631        if project_scope:
 632            storageProjects = list(set(storageProjects) & set(project_scope))
 633
 634            if not storageProjects:
 635                raise Warning(
 636                    f"There are no projects that the user has access to that match the criteria of the specified project scope: {project_scope}"
 637                )
 638
 639        # prepare a return list of project IDs and names
 640        projects = []
 641        for projectId in storageProjects:
 642            project_name_from_project_header = project_id_to_name_dict.get(projectId)
 643            projects.append((projectId, project_name_from_project_header))
 644
 645        sorted_projects_list = sorted(projects, key=lambda tup: tup[0])
 646
 647        return sorted_projects_list
 648
 649    @tracer.start_as_current_span("SynapseStorage::getStorageDatasetsInProject")
 650    def getStorageDatasetsInProject(self, projectId: str) -> list[tuple[str, str]]:
 651        """Gets all datasets in folder under a given storage project that the current user has access to.
 652
 653        Args:
 654            projectId: synapse ID of a storage project.
 655
 656        Returns:
 657            A list of datasets within the given storage project; the list consists of tuples (datasetId, datasetName).
 658            None: If the projectId cannot be found on Synapse.
 659        """
 660
 661        # select all folders and fetch their names from within the storage project;
 662        # if folder content type is defined, only select folders that contain datasets
 663        if "contentType" in self.storageFileviewTable.columns:
 664            foldersTable = self.storageFileviewTable[
 665                (self.storageFileviewTable["contentType"] == "dataset")
 666                & (self.storageFileviewTable["projectId"] == projectId)
 667            ]
 668        else:
 669            foldersTable = self.storageFileviewTable[
 670                (self.storageFileviewTable["type"] == "folder")
 671                & (self.storageFileviewTable["parentId"] == projectId)
 672            ]
 673
 674        # get an array of tuples (folderId, folderName)
 675        # some folders are part of datasets; others contain datasets
 676        # each dataset parent is the project; folders part of a dataset have another folder as a parent
 677        # to get folders if and only if they contain datasets for each folder
 678        # check if folder's parent is the project; if so that folder contains a dataset,
 679        # unless the folder list has already been filtered to dataset folders based on contentType attribute above
 680
 681        datasetList = []
 682        folderProperties = ["id", "name"]
 683        for folder in list(
 684            foldersTable[folderProperties].itertuples(index=False, name=None)
 685        ):
 686            datasetList.append(folder)
 687
 688        sorted_dataset_list = sorted(datasetList, key=lambda tup: tup[0])
 689
 690        return sorted_dataset_list
 691
 692    @tracer.start_as_current_span("SynapseStorage::getFilesInStorageDataset")
 693    def getFilesInStorageDataset(
 694        self, datasetId: str, fileNames: List = None, fullpath: bool = True
 695    ) -> List[Tuple[str, str]]:
 696        """Gets all files (excluding manifest files) in a given dataset folder.
 697
 698        Args:
 699            datasetId: synapse ID of a storage dataset.
 700            fileNames: get a list of files with particular names; defaults to None in which case all dataset files are returned (except bookkeeping files, e.g.
 701            metadata manifests); if fileNames is not None, all files matching the names in the fileNames list are returned if present.
 702            fullpath: if True return the full path as part of this filename; otherwise return just base filename
 703
 704        Returns:
 705            A list of files; the list consists of tuples (fileId, fileName).
 706
 707        Raises:
 708            ValueError: Dataset ID not found.
 709        """
 710        file_list = []
 711
 712        # Get path to dataset folder by using childern to avoid cases where the dataset is the scope of the view
 713        if self.storageFileviewTable.empty:
 714            raise ValueError(
 715                f"Fileview {self.storageFileview} is empty, please check the table and the provided synID and try again."
 716            )
 717
 718        child_path = self.storageFileviewTable.loc[
 719            self.storageFileviewTable["parentId"] == datasetId, "path"
 720        ]
 721        if child_path.empty:
 722            raise LookupError(
 723                f"Dataset {datasetId} could not be found in fileview {self.storageFileview}."
 724            )
 725        child_path = child_path.iloc[0]
 726
 727        # Get the dataset path by eliminating the child's portion of the path to account for nested datasets
 728        parent = child_path.split("/")[:-1]
 729        parent = "/".join(parent)
 730
 731        # Format dataset path to be used in table query
 732        dataset_path = f"'{parent}/%'"
 733
 734        # When querying, only include files to exclude entity files and subdirectories
 735        where_clauses = [f"path like {dataset_path}", "type='file'"]
 736
 737        # Requery the fileview to specifically get the files in the given dataset
 738        self.query_fileview(columns=["id", "path"], where_clauses=where_clauses)
 739
 740        # Exclude manifest files
 741        non_manifest_files = self.storageFileviewTable.loc[
 742            ~self.storageFileviewTable["path"].str.contains("synapse_storage_manifest"),
 743            :,
 744        ]
 745
 746        # Remove all files that are not in the list of fileNames
 747        if fileNames:
 748            filename_regex = "|".join(fileNames)
 749
 750            matching_files = non_manifest_files["path"].str.contains(
 751                filename_regex, case=False, regex=True
 752            )
 753
 754            non_manifest_files = non_manifest_files.loc[matching_files, :]
 755
 756        # Truncate path if necessary
 757        if not fullpath:
 758            non_manifest_files.path = non_manifest_files.path.apply(os.path.basename)
 759
 760        # Return list of files as expected by other methods
 761        file_list = list(non_manifest_files.itertuples(index=False, name=None))
 762
 763        return file_list
 764
 765    def _get_manifest_id(self, manifest: pd.DataFrame) -> str:
 766        """If both censored and uncensored manifests are present, return uncensored manifest; if only one manifest is present, return manifest id of that manifest; if more than two manifests are present, return the manifest id of the first one.
 767        Args:
 768        manifest: a dataframe contains name and id of manifests in a given asset view
 769
 770        Return:
 771        manifest_syn_id: id of a given censored or uncensored manifest
 772        """
 773        censored_regex = re.compile(".*censored.*")
 774        censored = manifest["name"].str.contains(censored_regex)
 775        if any(censored):
 776            # Try to use uncensored manifest first
 777            not_censored = ~censored
 778            if any(not_censored):
 779                manifest_syn_id = manifest[not_censored]["id"].iloc[0]
 780            # if only censored manifests are available, just use the first censored manifest
 781            else:
 782                manifest_syn_id = manifest["id"].iloc[0]
 783
 784        # otherwise, use the first (implied only) version that exists
 785        else:
 786            manifest_syn_id = manifest["id"].iloc[0]
 787
 788        return manifest_syn_id
 789
 790    @tracer.start_as_current_span("SynapseStorage::getDatasetManifest")
 791    def getDatasetManifest(
 792        self,
 793        datasetId: str,
 794        downloadFile: bool = False,
 795        newManifestName: str = "",
 796        use_temporary_folder: bool = True,
 797    ) -> Union[str, File]:
 798        """Gets the manifest associated with a given dataset.
 799
 800        Args:
 801            datasetId: synapse ID of a storage dataset.
 802            downloadFile: boolean argument indicating if manifest file in dataset should be downloaded or not.
 803            newManifestName: new name of a manifest that gets downloaded
 804            use_temporary_folder: boolean argument indicating if a temporary folder
 805                should be used to store the manifest file. This is useful when running
 806                this code as an API server where multiple requests could be made at the
 807                same time. This is set to False when the code is being used from the
 808                CLI. Defaults to True.
 809
 810        Returns:
 811            manifest_syn_id (String): Synapse ID of exisiting manifest file.
 812            manifest_data (synapseclient.entity.File): Synapse entity if downloadFile is True.
 813            "" (String): No pre-exisiting manifest in dataset.
 814        """
 815        manifest_data = ""
 816
 817        # get a list of files containing the manifest for this dataset (if any)
 818        all_files = self.storageFileviewTable
 819
 820        # construct regex based on manifest basename in the config
 821        manifest_re = re.compile(os.path.basename(self.manifest) + ".*.[tc]sv")
 822
 823        # search manifest based on given manifest basename regex above
 824        # and return a dataframe containing name and id of manifests in a given asset view
 825        manifest = all_files[
 826            (all_files["name"].str.contains(manifest_re, regex=True))
 827            & (all_files["parentId"] == datasetId)
 828        ]
 829
 830        manifest = manifest[["id", "name"]]
 831
 832        # if there is no pre-exisiting manifest in the specified dataset
 833        if manifest.empty:
 834            logger.warning(
 835                f"Could not find a manifest that fits basename {self.manifest} in asset view and dataset {datasetId}"
 836            )
 837            return ""
 838
 839        # if there is an exisiting manifest
 840        else:
 841            manifest_syn_id = self._get_manifest_id(manifest)
 842            if downloadFile:
 843                md = ManifestDownload(
 844                    self.syn,
 845                    manifest_id=manifest_syn_id,
 846                    synapse_entity_tracker=self.synapse_entity_tracker,
 847                )
 848                manifest_data = md.download_manifest(
 849                    newManifestName=newManifestName,
 850                    manifest_df=manifest,
 851                    use_temporary_folder=use_temporary_folder,
 852                )
 853                # TO DO: revisit how downstream code handle manifest_data. If the downstream code would break when manifest_data is an empty string,
 854                # then we should catch the error here without returning an empty string.
 855                if not manifest_data:
 856                    logger.debug(
 857                        f"No manifest data returned. Please check if you have successfully downloaded manifest: {manifest_syn_id}"
 858                    )
 859                return manifest_data
 860            return manifest_syn_id
 861
 862    def getDataTypeFromManifest(self, manifestId: str):
 863        """Fetch a manifest and return data types of all columns
 864        Args:
 865            manifestId: synapse ID of a manifest
 866        """
 867        # get manifest file path
 868        manifest_entity = self.synapse_entity_tracker.get(
 869            synapse_id=manifestId, syn=self.syn, download_file=True
 870        )
 871        manifest_filepath = manifest_entity.path
 872
 873        # load manifest dataframe
 874        manifest = load_df(
 875            manifest_filepath,
 876            preserve_raw_input=False,
 877            data_model=False,
 878        )
 879
 880        # convert the dataFrame to use best possible dtypes.
 881        manifest_new = manifest.convert_dtypes()
 882
 883        # get data types of columns
 884        result = manifest_new.dtypes.to_frame("dtypes").reset_index()
 885
 886        # return the result as a dictionary
 887        result_dict = result.set_index("index")["dtypes"].astype(str).to_dict()
 888
 889        return result_dict
 890
 891    def _get_files_metadata_from_dataset(
 892        self, datasetId: str, only_new_files: bool, manifest: pd.DataFrame = None
 893    ) -> Optional[dict]:
 894        """retrieve file ids under a particular datasetId
 895
 896        Args:
 897            datasetId (str): a dataset id
 898            only_new_files (bool): if only adding new files that are not already exist
 899            manifest (pd.DataFrame): metadata manifest dataframe. Default to None.
 900
 901        Returns:
 902            a dictionary that contains filename and entityid under a given datasetId or None if there is nothing under a given dataset id are not available
 903        """
 904        dataset_files = self.getFilesInStorageDataset(datasetId)
 905        if dataset_files:
 906            dataset_file_names_id_dict = self._get_file_entityIds(
 907                dataset_files, only_new_files=only_new_files, manifest=manifest
 908            )
 909            return dataset_file_names_id_dict
 910        else:
 911            return None
 912
 913    def add_entity_id_and_filename(
 914        self, datasetId: str, manifest: pd.DataFrame
 915    ) -> pd.DataFrame:
 916        """add entityid and filename column to an existing manifest assuming entityId column is not already present
 917
 918        Args:
 919            datasetId (str): dataset syn id
 920            manifest (pd.DataFrame): existing manifest dataframe, assuming this dataframe does not have an entityId column and Filename column is present but completely empty
 921
 922        Returns:
 923            pd.DataFrame: returns a pandas dataframe
 924        """
 925        # get file names and entity ids of a given dataset
 926        dataset_files_dict = self._get_files_metadata_from_dataset(
 927            datasetId, only_new_files=False
 928        )
 929
 930        if dataset_files_dict:
 931            # turn manifest dataframe back to a dictionary for operation
 932            manifest_dict = manifest.to_dict("list")
 933
 934            # update Filename column
 935            # add entityId column to the end
 936            manifest_dict.update(dataset_files_dict)
 937
 938            # if the component column exists in existing manifest, fill up that column
 939            if "Component" in manifest_dict.keys():
 940                manifest_dict["Component"] = manifest_dict["Component"] * max(
 941                    1, len(manifest_dict["Filename"])
 942                )
 943
 944            # turn dictionary back to a dataframe
 945            manifest_df_index = pd.DataFrame.from_dict(manifest_dict, orient="index")
 946            manifest_df_updated = manifest_df_index.transpose()
 947
 948            # fill na with empty string
 949            manifest_df_updated = manifest_df_updated.fillna("")
 950
 951            # drop index
 952            manifest_df_updated = manifest_df_updated.reset_index(drop=True)
 953
 954            return manifest_df_updated
 955        else:
 956            return manifest
 957
 958    def fill_in_entity_id_filename(
 959        self, datasetId: str, manifest: pd.DataFrame
 960    ) -> Tuple[List, pd.DataFrame]:
 961        """fill in Filename column and EntityId column. EntityId column and Filename column will be created if not already present.
 962
 963        Args:
 964            datasetId (str): dataset syn id
 965            manifest (pd.DataFrame): existing manifest dataframe.
 966
 967        Returns:
 968            Tuple[List, pd.DataFrame]: a list of synIds that are under a given datasetId folder and updated manifest dataframe
 969        """
 970        # get dataset file names and entity id as a list of tuple
 971        dataset_files = self.getFilesInStorageDataset(datasetId)
 972
 973        # update manifest with additional filenames, if any
 974        # note that if there is an existing manifest and there are files in the dataset
 975        # the columns Filename and entityId are assumed to be present in manifest schema
 976        # TODO: use idiomatic panda syntax
 977        if not dataset_files:
 978            manifest = manifest.fillna("")
 979            return dataset_files, manifest
 980
 981        all_files = self._get_file_entityIds(
 982            dataset_files=dataset_files, only_new_files=False, manifest=manifest
 983        )
 984        new_files = self._get_file_entityIds(
 985            dataset_files=dataset_files, only_new_files=True, manifest=manifest
 986        )
 987
 988        all_files = pd.DataFrame(all_files)
 989        new_files = pd.DataFrame(new_files)
 990
 991        # update manifest so that it contains new dataset files
 992        manifest = (
 993            pd.concat([manifest, new_files], sort=False)
 994            .reset_index()
 995            .drop("index", axis=1)
 996        )
 997
 998        # Reindex manifest and new files dataframes according to entityIds to align file paths and metadata
 999        manifest_reindex = manifest.set_index("entityId")
1000        all_files_reindex = all_files.set_index("entityId")
1001        all_files_reindex_like_manifest = all_files_reindex.reindex_like(
1002            manifest_reindex
1003        )
1004
1005        # Check if individual file paths in manifest and from synapse match
1006        file_paths_match = (
1007            manifest_reindex["Filename"] == all_files_reindex_like_manifest["Filename"]
1008        )
1009
1010        # If all the paths do not match, update the manifest with the filepaths from synapse
1011        if not file_paths_match.all():
1012            manifest_reindex.loc[
1013                ~file_paths_match, "Filename"
1014            ] = all_files_reindex_like_manifest.loc[~file_paths_match, "Filename"]
1015
1016            # reformat manifest for further use
1017            manifest = manifest_reindex.reset_index()
1018            entityIdCol = manifest.pop("entityId")
1019            manifest.insert(len(manifest.columns), "entityId", entityIdCol)
1020
1021        manifest = manifest.fillna("")
1022        return dataset_files, manifest
1023
1024    @tracer.start_as_current_span("SynapseStorage::updateDatasetManifestFiles")
1025    def updateDatasetManifestFiles(
1026        self, dmge: DataModelGraphExplorer, datasetId: str, store: bool = True
1027    ) -> Union[Tuple[str, pd.DataFrame], None]:
1028        """Fetch the names and entity IDs of all current files in dataset in store, if any; update dataset's manifest with new files, if any.
1029
1030        Args:
1031            dmge: DataModelGraphExplorer Instance
1032            datasetId: synapse ID of a storage dataset.
1033            store: if set to True store updated manifest in asset store; if set to False
1034            return a Pandas dataframe containing updated manifest but do not store to asset store
1035
1036
1037        Returns:
1038            Synapse ID of updated manifest and Pandas dataframe containing the updated manifest.
1039            If there is no existing manifest or if the manifest does not have an entityId column, return None
1040        """
1041
1042        # get existing manifest Synapse ID
1043        manifest_id = self.getDatasetManifest(datasetId)
1044
1045        # if there is no manifest return None
1046        if not manifest_id:
1047            return None
1048
1049        manifest_entity = self.synapse_entity_tracker.get(
1050            synapse_id=manifest_id, syn=self.syn, download_file=True
1051        )
1052        manifest_filepath = manifest_entity.path
1053        manifest = load_df(manifest_filepath)
1054
1055        # If the manifest does not have an entityId column, trigger a new manifest to be generated
1056        if "entityId" not in manifest.columns:
1057            return None
1058
1059        manifest_is_file_based = "Filename" in manifest.columns
1060
1061        if manifest_is_file_based:
1062            # update manifest with additional filenames, if any
1063            # note that if there is an existing manifest and there are files in the dataset
1064            # the columns Filename and entityId are assumed to be present in manifest schema
1065            # TODO: use idiomatic panda syntax
1066            dataset_files, manifest = self.fill_in_entity_id_filename(
1067                datasetId, manifest
1068            )
1069            if dataset_files:
1070                # update the manifest file, so that it contains the relevant entity IDs
1071                if store:
1072                    manifest.to_csv(manifest_filepath, index=False)
1073
1074                    # store manifest and update associated metadata with manifest on Synapse
1075                    manifest_id = self.associateMetadataWithFiles(
1076                        dmge, manifest_filepath, datasetId
1077                    )
1078
1079        return manifest_id, manifest
1080
1081    def _get_file_entityIds(
1082        self,
1083        dataset_files: List,
1084        only_new_files: bool = False,
1085        manifest: pd.DataFrame = None,
1086    ):
1087        """
1088        Get a dictionary of files in a dataset. Either files that are not in the current manifest or all files
1089
1090        Args:
1091            manifest: metadata manifest
1092            dataset_file: List of all files in a dataset
1093            only_new_files: boolean to control whether only new files are returned or all files in the dataset
1094        Returns:
1095            files: dictionary of file names and entityIDs, with scope as specified by `only_new_files`
1096        """
1097        files = {"Filename": [], "entityId": []}
1098
1099        if only_new_files:
1100            if manifest is None:
1101                raise UnboundLocalError(
1102                    "No manifest was passed in, a manifest is required when `only_new_files` is True."
1103                )
1104
1105            if "entityId" not in manifest.columns:
1106                raise ValueError(
1107                    "The manifest in your dataset and/or top level folder must contain the 'entityId' column. "
1108                    "Please generate an empty manifest without annotations, manually add annotations to the "
1109                    "appropriate files in the manifest, and then try again."
1110                )
1111
1112            # find new files (that are not in the current manifest) if any
1113            for file_id, file_name in dataset_files:
1114                if not file_id in manifest["entityId"].values:
1115                    files["Filename"].append(file_name)
1116                    files["entityId"].append(file_id)
1117        else:
1118            # get all files
1119            for file_id, file_name in dataset_files:
1120                files["Filename"].append(file_name)
1121                files["entityId"].append(file_id)
1122
1123        return files
1124
1125    @tracer.start_as_current_span("SynapseStorage::getProjectManifests")
1126    def getProjectManifests(
1127        self, projectId: str
1128    ) -> list[tuple[tuple[str, str], tuple[str, str], tuple[str, str]]]:
1129        """Gets all metadata manifest files across all datasets in a specified project.
1130
1131        Returns: A list of datasets per project; metadata manifest Synapse ID for each dataset; and the corresponding schema component of the manifest
1132                 as a list of tuples, one for each manifest:
1133                    [
1134                        (
1135                            (datasetId, dataName),
1136                            (manifestId, manifestName),
1137                            (componentSchemaLabel, componentSchemaLabel) TODO: # get component name from schema
1138                        ),
1139                        ...
1140                    ]
1141
1142        TODO: Return manifest URI instead of Synapse ID for interoperability with other implementations of a store interface
1143        """
1144        component = None
1145        entity = None
1146        manifests = []
1147
1148        datasets = self.getStorageDatasetsInProject(projectId)
1149
1150        for datasetId, datasetName in datasets:
1151            # encode information about the manifest in a simple list (so that R clients can unpack it)
1152            # eventually can serialize differently
1153
1154            # Get synID of manifest for a dataset
1155            manifestId = self.getDatasetManifest(datasetId)
1156
1157            # If a manifest exists, get the annotations for it, else return base 'manifest' tuple
1158            if manifestId:
1159                annotations = self.getFileAnnotations(manifestId)
1160
1161                # If manifest has annotations specifying component, use that
1162                if annotations and "Component" in annotations:
1163                    component = annotations["Component"]
1164                    entity = self.synapse_entity_tracker.get(
1165                        synapse_id=manifestId, syn=self.syn, download_file=False
1166                    )
1167                    manifest_name = entity["properties"]["name"]
1168
1169                # otherwise download the manifest and parse for information
1170                elif not annotations or "Component" not in annotations:
1171                    logging.debug(
1172                        f"No component annotations have been found for manifest {manifestId}. "
1173                        "The manifest will be downloaded and parsed instead. "
1174                        "For increased speed, add component annotations to manifest."
1175                    )
1176
1177                    manifest_info = self.getDatasetManifest(
1178                        datasetId, downloadFile=True
1179                    )
1180                    manifest_name = manifest_info["properties"].get("name", "")
1181
1182                    if not manifest_name:
1183                        logger.error(f"Failed to download manifests from {datasetId}")
1184
1185                    manifest_path = manifest_info["path"]
1186
1187                    manifest_df = load_df(manifest_path)
1188
1189                    # Get component from component column if it exists
1190                    if (
1191                        "Component" in manifest_df
1192                        and not manifest_df["Component"].empty
1193                    ):
1194                        list(set(manifest_df["Component"]))
1195                        component = list(set(manifest_df["Component"]))
1196
1197                        # Added to address issues raised during DCA testing
1198                        if "" in component:
1199                            component.remove("")
1200
1201                        if len(component) == 1:
1202                            component = component[0]
1203                        elif len(component) > 1:
1204                            logging.warning(
1205                                f"Manifest {manifestId} is composed of multiple components. Schematic does not support mulit-component manifests at this time."
1206                                "Behavior of manifests with multiple components is undefined"
1207                            )
1208            else:
1209                manifest_name = ""
1210                component = None
1211            if component:
1212                manifest = (
1213                    (datasetId, datasetName),
1214                    (manifestId, manifest_name),
1215                    (component, component),
1216                )
1217            elif manifestId:
1218                logging.debug(
1219                    f"Manifest {manifestId} does not have an associated Component"
1220                )
1221                manifest = (
1222                    (datasetId, datasetName),
1223                    (manifestId, manifest_name),
1224                    ("", ""),
1225                )
1226            else:
1227                manifest = (
1228                    (datasetId, datasetName),
1229                    ("", ""),
1230                    ("", ""),
1231                )
1232
1233            if manifest:
1234                manifests.append(manifest)
1235
1236        return manifests
1237
1238    def upload_project_manifests_to_synapse(
1239        self, dmge: DataModelGraphExplorer, projectId: str
1240    ) -> List[str]:
1241        """Upload all metadata manifest files across all datasets in a specified project as tables in Synapse.
1242
1243        Returns: String of all the manifest_table_ids of all the manifests that have been loaded.
1244        """
1245
1246        manifests = []
1247        manifest_loaded = []
1248        datasets = self.getStorageDatasetsInProject(projectId)
1249
1250        for datasetId, datasetName in datasets:
1251            # encode information about the manifest in a simple list (so that R clients can unpack it)
1252            # eventually can serialize differently
1253
1254            manifest = ((datasetId, datasetName), ("", ""), ("", ""))
1255
1256            manifest_info = self.getDatasetManifest(datasetId, downloadFile=True)
1257            if manifest_info:
1258                manifest_id = manifest_info["properties"]["id"]
1259                manifest_name = manifest_info["properties"]["name"]
1260                manifest_path = manifest_info["path"]
1261                manifest_df = load_df(manifest_path)
1262                manifest_table_id = uploadDB(
1263                    dmge=dmge,
1264                    manifest=manifest,
1265                    datasetId=datasetId,
1266                    table_name=datasetName,
1267                )
1268                manifest_loaded.append(datasetName)
1269        return manifest_loaded
1270
1271    def upload_annotated_project_manifests_to_synapse(
1272        self, projectId: str, path_to_json_ld: str, dry_run: bool = False
1273    ) -> List[str]:
1274        """
1275        Purpose:
1276            For all manifests in a project, upload them as a table and add annotations manifest csv.
1277            Assumes the manifest is already present as a CSV in a dataset in the project.
1278
1279        """
1280        # Instantiate DataModelParser
1281        data_model_parser = DataModelParser(path_to_data_model=path_to_json_ld)
1282        # Parse Model
1283        parsed_data_model = data_model_parser.parse_model()
1284
1285        # Instantiate DataModelGraph
1286        data_model_grapher = DataModelGraph(parsed_data_model)
1287
1288        # Generate graph
1289        graph_data_model = data_model_grapher.generate_data_model_graph()
1290
1291        # Instantiate DataModelGraphExplorer
1292        dmge = DataModelGraphExplorer(graph_data_model)
1293
1294        manifests = []
1295        manifest_loaded = []
1296        datasets = self.getStorageDatasetsInProject(projectId)
1297        for datasetId, datasetName in datasets:
1298            # encode information about the manifest in a simple list (so that R clients can unpack it)
1299            # eventually can serialize differently
1300
1301            manifest = ((datasetId, datasetName), ("", ""), ("", ""))
1302            manifests.append(manifest)
1303
1304            manifest_info = self.getDatasetManifest(datasetId, downloadFile=True)
1305
1306            if manifest_info:
1307                manifest_id = manifest_info["properties"]["id"]
1308                manifest_name = manifest_info["properties"]["name"]
1309                manifest_path = manifest_info["path"]
1310                manifest = (
1311                    (datasetId, datasetName),
1312                    (manifest_id, manifest_name),
1313                    ("", ""),
1314                )
1315                if not dry_run:
1316                    self.associateMetadataWithFiles(
1317                        dmge, manifest_path, datasetId, manifest_record_type="table"
1318                    )
1319                manifest_loaded.append(manifest)
1320
1321        return manifests, manifest_loaded
1322
1323    def move_entities_to_new_project(
1324        self,
1325        projectId: str,
1326        newProjectId: str,
1327        returnEntities: bool = False,
1328        dry_run: bool = False,
1329    ):
1330        """
1331        For each manifest csv in a project, look for all the entitiy ids that are associated.
1332        Look up the entitiy in the files, move the entity to new project.
1333        """
1334
1335        manifests = []
1336        manifest_loaded = []
1337        datasets = self.getStorageDatasetsInProject(projectId)
1338        if datasets:
1339            for datasetId, datasetName in datasets:
1340                # encode information about the manifest in a simple list (so that R clients can unpack it)
1341                # eventually can serialize differently
1342
1343                manifest = ((datasetId, datasetName), ("", ""), ("", ""))
1344                manifests.append(manifest)
1345
1346                manifest_info = self.getDatasetManifest(datasetId, downloadFile=True)
1347                if manifest_info:
1348                    manifest_id = manifest_info["properties"]["id"]
1349                    manifest_name = manifest_info["properties"]["name"]
1350                    manifest_path = manifest_info["path"]
1351                    manifest_df = load_df(manifest_path)
1352
1353                    manifest = (
1354                        (datasetId, datasetName),
1355                        (manifest_id, manifest_name),
1356                        ("", ""),
1357                    )
1358                    manifest_loaded.append(manifest)
1359
1360                    annotation_entities = self.storageFileviewTable[
1361                        (self.storageFileviewTable["id"].isin(manifest_df["entityId"]))
1362                        & (self.storageFileviewTable["type"] == "folder")
1363                    ]["id"]
1364
1365                    if returnEntities:
1366                        for entityId in annotation_entities:
1367                            if not dry_run:
1368                                moved_entity = self.syn.move(entityId, datasetId)
1369                                self.synapse_entity_tracker.add(
1370                                    synapse_id=moved_entity.id, entity=moved_entity
1371                                )
1372                            else:
1373                                logging.info(
1374                                    f"{entityId} will be moved to folder {datasetId}."
1375                                )
1376                    else:
1377                        # generate project folder
1378                        archive_project_folder = Folder(
1379                            projectId + "_archive", parent=newProjectId
1380                        )
1381                        archive_project_folder = self.syn.store(archive_project_folder)
1382                        self.synapse_entity_tracker.add(
1383                            synapse_id=archive_project_folder.id,
1384                            entity=archive_project_folder,
1385                        )
1386
1387                        # generate dataset folder
1388                        dataset_archive_folder = Folder(
1389                            "_".join([datasetId, datasetName, "archive"]),
1390                            parent=archive_project_folder.id,
1391                        )
1392                        dataset_archive_folder = self.syn.store(dataset_archive_folder)
1393                        self.synapse_entity_tracker.add(
1394                            synapse_id=dataset_archive_folder.id,
1395                            entity=dataset_archive_folder,
1396                        )
1397
1398                        for entityId in annotation_entities:
1399                            # move entities to folder
1400                            if not dry_run:
1401                                moved_entity = self.syn.move(
1402                                    entityId, dataset_archive_folder.id
1403                                )
1404                                self.synapse_entity_tracker.add(
1405                                    synapse_id=moved_entity.id, entity=moved_entity
1406                                )
1407                            else:
1408                                logging.info(
1409                                    f"{entityId} will be moved to folder {dataset_archive_folder.id}."
1410                                )
1411        else:
1412            raise LookupError(
1413                f"No datasets were found in the specified project: {projectId}. Re-check specified master_fileview in CONFIG and retry."
1414            )
1415        return manifests, manifest_loaded
1416
1417    @tracer.start_as_current_span("SynapseStorage::get_synapse_table")
1418    def get_synapse_table(self, synapse_id: str) -> Tuple[pd.DataFrame, CsvFileTable]:
1419        """Download synapse table as a pd dataframe; return table schema and etags as results too
1420
1421        Args:
1422            synapse_id: synapse ID of the table to query
1423        """
1424
1425        results = self.syn.tableQuery("SELECT * FROM {}".format(synapse_id))
1426        df = results.asDataFrame(
1427            rowIdAndVersionInIndex=False,
1428            na_values=STR_NA_VALUES_FILTERED,
1429            keep_default_na=False,
1430        )
1431
1432        return df, results
1433
1434    @missing_entity_handler
1435    @tracer.start_as_current_span("SynapseStorage::uploadDB")
1436    def uploadDB(
1437        self,
1438        dmge: DataModelGraphExplorer,
1439        manifest: pd.DataFrame,
1440        datasetId: str,
1441        table_name: str,
1442        restrict: bool = False,
1443        table_manipulation: str = "replace",
1444        table_column_names: str = "class_label",
1445    ):
1446        """
1447        Method to upload a database to an asset store. In synapse, this will upload a metadata table
1448
1449        Args:
1450            dmge: DataModelGraphExplorer object
1451            manifest: pd.Df manifest to upload
1452            datasetId: synID of the dataset for the manifest
1453            table_name: name of the table to be uploaded
1454            restrict: bool, whether or not the manifest contains sensitive data that will need additional access restrictions
1455            existingTableId: str of the synId of the existing table, if one already exists
1456            table_manipulation: str, 'replace' or 'upsert', in the case where a manifest already exists, should the new metadata replace the existing (replace) or be added to it (upsert)
1457            table_column_names: (str): display_name/display_label/class_label (default). Sets labeling style for table column names. display_name will use the raw display name as the column name. class_label will format the display
1458                name as upper camelcase, and strip blacklisted characters, display_label will strip blacklisted characters including spaces, to retain
1459                display label formatting.
1460        Returns:
1461            manifest_table_id: synID of the uploaded table
1462            manifest: the original manifset
1463            table_manifest: manifest formatted appropriately for the table
1464
1465        """
1466
1467        col_schema, table_manifest = self.formatDB(
1468            dmge=dmge, manifest=manifest, table_column_names=table_column_names
1469        )
1470
1471        manifest_table_id = self.buildDB(
1472            datasetId,
1473            table_name,
1474            col_schema,
1475            table_manifest,
1476            table_manipulation,
1477            dmge,
1478            restrict,
1479        )
1480
1481        return manifest_table_id, manifest, table_manifest
1482
1483    @tracer.start_as_current_span("SynapseStorage::formatDB")
1484    def formatDB(self, dmge, manifest, table_column_names):
1485        """
1486        Method to format a manifest appropriatly for upload as table
1487
1488        Args:
1489            dmge: DataModelGraphExplorer object
1490            manifest: pd.Df manifest to upload
1491            table_column_names: (str): display_name/display_label/class_label (default). Sets labeling style for table column names. display_name will use the raw display name as the column name. class_label will format the display
1492                name as upper camelcase, and strip blacklisted characters, display_label will strip blacklisted characters including spaces, to retain
1493                display label formatting.
1494        Returns:
1495            col_schema: schema for table columns: type, size, etc
1496            table_manifest: formatted manifest
1497
1498        """
1499        # Rename the manifest columns to display names to match fileview
1500
1501        blacklist_chars = ["(", ")", ".", " ", "-"]
1502        manifest_columns = manifest.columns.tolist()
1503
1504        table_manifest = deepcopy(manifest)
1505
1506        if table_column_names == "display_name":
1507            cols = table_manifest.columns
1508
1509        elif table_column_names == "display_label":
1510            cols = [
1511                str(col).translate({ord(x): "" for x in blacklist_chars})
1512                for col in manifest_columns
1513            ]
1514
1515        elif table_column_names == "class_label":
1516            cols = [
1517                get_class_label_from_display_name(str(col)).translate(
1518                    {ord(x): "" for x in blacklist_chars}
1519                )
1520                for col in manifest_columns
1521            ]
1522        else:
1523            ValueError(
1524                f"The provided table_column_name: {table_column_names} is not valid, please resubmit with an allowed value only."
1525            )
1526
1527        cols = list(map(lambda x: x.replace("EntityId", "entityId"), cols))
1528
1529        # Reset column names in table manifest
1530        table_manifest.columns = cols
1531
1532        # move entity id to end of df
1533        entity_col = table_manifest.pop("entityId")
1534        table_manifest.insert(len(table_manifest.columns), "entityId", entity_col)
1535
1536        # Get the column schema
1537        col_schema = as_table_columns(table_manifest)
1538
1539        # Set Id column length to 64 (for some reason not being auto set.)
1540        for i, col in enumerate(col_schema):
1541            if col["name"].lower() == "id":
1542                col_schema[i]["maximumSize"] = 64
1543
1544        return col_schema, table_manifest
1545
1546    @tracer.start_as_current_span("SynapseStorage::buildDB")
1547    def buildDB(
1548        self,
1549        datasetId: str,
1550        table_name: str,
1551        col_schema: List,
1552        table_manifest: pd.DataFrame,
1553        table_manipulation: str,
1554        dmge: DataModelGraphExplorer,
1555        restrict: bool = False,
1556    ):
1557        """
1558        Method to construct the table appropriately: create new table, replace existing, or upsert new into existing
1559        Calls TableOperations class to execute
1560
1561        Args:
1562            datasetId: synID of the dataset for the manifest
1563            table_name: name of the table to be uploaded
1564            col_schema: schema for table columns: type, size, etc from `formatDB`
1565            table_manifest: formatted manifest that can be uploaded as a table
1566            table_manipulation: str, 'replace' or 'upsert', in the case where a manifest already exists, should the new metadata replace the existing (replace) or be added to it (upsert)
1567            restrict: bool, whether or not the manifest contains sensitive data that will need additional access restrictions
1568
1569        Returns:
1570            manifest_table_id: synID of the uploaded table
1571
1572        """
1573        table_parent_id = self.getDatasetProject(datasetId=datasetId)
1574        existing_table_id = self.syn.findEntityId(
1575            name=table_name, parent=table_parent_id
1576        )
1577
1578        tableOps = TableOperations(
1579            synStore=self,
1580            tableToLoad=table_manifest,
1581            tableName=table_name,
1582            datasetId=datasetId,
1583            existingTableId=existing_table_id,
1584            restrict=restrict,
1585            synapse_entity_tracker=self.synapse_entity_tracker,
1586        )
1587
1588        if not table_manipulation or existing_table_id is None:
1589            manifest_table_id = tableOps.createTable(
1590                columnTypeDict=col_schema,
1591                specifySchema=True,
1592            )
1593        elif existing_table_id is not None:
1594            if table_manipulation.lower() == "replace":
1595                manifest_table_id = tableOps.replaceTable(
1596                    specifySchema=True,
1597                    columnTypeDict=col_schema,
1598                )
1599            elif table_manipulation.lower() == "upsert":
1600                manifest_table_id = tableOps.upsertTable(
1601                    dmge=dmge,
1602                )
1603            elif table_manipulation.lower() == "update":
1604                manifest_table_id = tableOps.updateTable()
1605
1606        if table_manipulation and table_manipulation.lower() == "upsert":
1607            table_entity = self.synapse_entity_tracker.get(
1608                synapse_id=existing_table_id or manifest_table_id,
1609                syn=self.syn,
1610                download_file=False,
1611            )
1612            annos = OldAnnotations(
1613                id=table_entity.id,
1614                etag=table_entity.etag,
1615                values=table_entity.annotations,
1616            )
1617            annos["primary_key"] = table_manifest["Component"][0] + "_id"
1618            annos = self.syn.set_annotations(annos)
1619            table_entity.etag = annos.etag
1620            table_entity.annotations = annos
1621
1622        return manifest_table_id
1623
1624    @tracer.start_as_current_span("SynapseStorage::upload_manifest_file")
1625    def upload_manifest_file(
1626        self,
1627        manifest,
1628        metadataManifestPath,
1629        datasetId,
1630        restrict_manifest,
1631        component_name="",
1632    ):
1633        # Update manifest to have the new entityId column
1634        manifest.to_csv(metadataManifestPath, index=False)
1635
1636        # store manifest to Synapse as a CSV
1637        # update file name
1638        file_name_full = metadataManifestPath.split("/")[-1]
1639        file_extension = file_name_full.split(".")[-1]
1640
1641        # Differentiate "censored" and "uncensored" manifest
1642        if "censored" in file_name_full:
1643            file_name_new = (
1644                os.path.basename(CONFIG.synapse_manifest_basename)
1645                + "_"
1646                + component_name
1647                + "_censored"
1648                + "."
1649                + file_extension
1650            )
1651        else:
1652            file_name_new = (
1653                os.path.basename(CONFIG.synapse_manifest_basename)
1654                + "_"
1655                + component_name
1656                + "."
1657                + file_extension
1658            )
1659
1660        manifest_synapse_file = None
1661        try:
1662            # Rename the file to file_name_new then revert
1663            # This is to maintain the original file name in-case other code is
1664            # expecting that the file exists with the original name
1665            original_file_path = metadataManifestPath
1666            new_file_path = os.path.join(
1667                os.path.dirname(metadataManifestPath), file_name_new
1668            )
1669            os.rename(original_file_path, new_file_path)
1670
1671            manifest_synapse_file = self._store_file_for_manifest_upload(
1672                new_file_path=new_file_path,
1673                dataset_id=datasetId,
1674                existing_file_name=file_name_full,
1675                file_name_new=file_name_new,
1676                restrict_manifest=restrict_manifest,
1677            )
1678            manifest_synapse_file_id = manifest_synapse_file.id
1679
1680        finally:
1681            # Revert the file name back to the original
1682            os.rename(new_file_path, original_file_path)
1683
1684            if manifest_synapse_file:
1685                manifest_synapse_file.path = original_file_path
1686
1687        return manifest_synapse_file_id
1688
1689    def _store_file_for_manifest_upload(
1690        self,
1691        new_file_path: str,
1692        dataset_id: str,
1693        existing_file_name: str,
1694        file_name_new: str,
1695        restrict_manifest: bool,
1696    ) -> File:
1697        """Handles a create or update of a manifest file that is going to be uploaded.
1698        If we already have a copy of the Entity in memory we will update that instance,
1699        otherwise create a new File instance to be created in Synapse. Once stored
1700        this will add the file to the `synapse_entity_tracker` for future reference.
1701
1702        Args:
1703            new_file_path (str): The path to the new manifest file
1704            dataset_id (str): The Synapse ID of the dataset the manifest is associated with
1705            existing_file_name (str): The name of the existing file
1706            file_name_new (str): The name of the new file
1707            restrict_manifest (bool): Whether the manifest should be restricted
1708
1709        Returns:
1710            File: The stored manifest file
1711        """
1712        local_tracked_file_instance = (
1713            self.synapse_entity_tracker.search_local_by_parent_and_name(
1714                name=existing_file_name, parent_id=dataset_id
1715            )
1716            or self.synapse_entity_tracker.search_local_by_parent_and_name(
1717                name=file_name_new, parent_id=dataset_id
1718            )
1719        )
1720
1721        if local_tracked_file_instance:
1722            local_tracked_file_instance.path = new_file_path
1723            local_tracked_file_instance.description = (
1724                "Manifest for dataset " + dataset_id
1725            )
1726            manifest_synapse_file = local_tracked_file_instance
1727        else:
1728            manifest_synapse_file = File(
1729                path=new_file_path,
1730                description="Manifest for dataset " + dataset_id,
1731                parent=dataset_id,
1732                name=file_name_new,
1733            )
1734
1735        manifest_synapse_file = self.syn.store(
1736            manifest_synapse_file, isRestricted=restrict_manifest
1737        )
1738
1739        self.synapse_entity_tracker.add(
1740            synapse_id=manifest_synapse_file.id, entity=manifest_synapse_file
1741        )
1742        return manifest_synapse_file
1743
1744    async def get_async_annotation(self, synapse_id: str) -> Dict[str, Any]:
1745        """get annotations asynchronously
1746
1747        Args:
1748            synapse_id (str): synapse id of the entity that the annotation belongs
1749
1750        Returns:
1751            Dict[str, Any]: The requested entity bundle matching
1752            <https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/entitybundle/v2/EntityBundle.html>
1753        """
1754        return await get_entity_id_bundle2(
1755            entity_id=synapse_id,
1756            request={"includeAnnotations": True},
1757            synapse_client=self.syn,
1758        )
1759
1760    async def store_async_annotation(self, annotation_dict: dict) -> Annotations:
1761        """store annotation in an async way
1762
1763        Args:
1764            annotation_dict (dict): annotation in a dictionary format
1765
1766        Returns:
1767            Annotations: The stored annotations.
1768        """
1769        annotation_data = Annotations.from_dict(
1770            synapse_annotations=annotation_dict["annotations"]["annotations"]
1771        )
1772        annotation_class = Annotations(
1773            annotations=annotation_data,
1774            etag=annotation_dict["annotations"]["etag"],
1775            id=annotation_dict["annotations"]["id"],
1776        )
1777        annotation_storage_result = await annotation_class.store_async(
1778            synapse_client=self.syn
1779        )
1780        local_entity = self.synapse_entity_tracker.get(
1781            synapse_id=annotation_dict["annotations"]["id"],
1782            syn=self.syn,
1783            download_file=False,
1784            retrieve_if_not_present=False,
1785        )
1786        if local_entity:
1787            local_entity.etag = annotation_storage_result.etag
1788            local_entity.annotations = annotation_storage_result
1789        return annotation_storage_result
1790
1791    def process_row_annotations(
1792        self,
1793        dmge: DataModelGraphExplorer,
1794        metadata_syn: Dict[str, Any],
1795        hide_blanks: bool,
1796        csv_list_regex: str,
1797        annos: Dict[str, Any],
1798        annotation_keys: str,
1799    ) -> Dict[str, Any]:
1800        """Processes metadata annotations based on the logic below:
1801        1. Checks if the hide_blanks flag is True, and if the current annotation value (anno_v) is:
1802            An empty or whitespace-only string.
1803            A NaN value (if the annotation is a float).
1804        if any of the above conditions are met, and hide_blanks is True, the annotation key is not going to be uploaded and skips further processing of that annotation key.
1805        if any of the above conditions are met, and hide_blanks is False, assigns an empty string "" as the annotation value for that key.
1806
1807        2. If the value is a string and matches the pattern defined by csv_list_regex, get validation rule based on "node label" or "node display name".
1808        Check if the rule contains "list" as a rule, if it does, split the string by comma and assign the resulting list as the annotation value for that key.
1809
1810        3. For any other conditions, assigns the original value of anno_v to the annotation key (anno_k).
1811
1812        4. Returns the updated annotations dictionary.
1813
1814        Args:
1815            dmge (DataModelGraphExplorer): data model graph explorer
1816            metadata_syn (dict): metadata used for Synapse storage
1817            hideBlanks (bool): if true, does not upload annotation keys with blank values.
1818            csv_list_regex (str): Regex to match with comma separated list
1819            annos (Dict[str, Any]): dictionary of annotation returned from synapse
1820            annotation_keys (str): display_label/class_label
1821
1822        Returns:
1823            Dict[str, Any]: annotations as a dictionary
1824
1825        ```mermaid
1826        flowchart TD
1827            A[Start] --> C{Is anno_v empty, whitespace, or NaN?}
1828            C -- Yes --> D{Is hide_blanks True?}
1829            D -- Yes --> E[Remove this annotation key from the annotation dictionary to be uploaded. Skip further processing]
1830            D -- No --> F[Assign empty string to annotation key]
1831            C -- No --> G{Is anno_v a string?}
1832            G -- No --> H[Assign original value of anno_v to annotation key]
1833            G -- Yes --> I{Does anno_v match csv_list_regex?}
1834            I -- Yes --> J[Get validation rule of anno_k]
1835            J --> K{Does the validation rule contain 'list'}
1836            K -- Yes --> L[Split anno_v by commas and assign as list]
1837            I -- No --> H
1838            K -- No --> H
1839        ```
1840        """
1841        for anno_k, anno_v in metadata_syn.items():
1842            # Remove keys with nan or empty string values or string that only contains white space from dict of annotations to be uploaded
1843            # if present on current data annotation
1844            if hide_blanks and (
1845                (isinstance(anno_v, str) and anno_v.strip() == "")
1846                or (isinstance(anno_v, float) and np.isnan(anno_v))
1847            ):
1848                annos["annotations"]["annotations"].pop(anno_k) if anno_k in annos[
1849                    "annotations"
1850                ]["annotations"].keys() else annos["annotations"]["annotations"]
1851                continue
1852
1853            # Otherwise save annotation as approrpriate
1854            if isinstance(anno_v, float) and np.isnan(anno_v):
1855                annos["annotations"]["annotations"][anno_k] = ""
1856                continue
1857
1858            # Handle strings that match the csv_list_regex and pass the validation rule
1859            if isinstance(anno_v, str) and re.fullmatch(csv_list_regex, anno_v):
1860                # Use a dictionary to dynamically choose the argument
1861                param = (
1862                    {"node_display_name": anno_k}
1863                    if annotation_keys == "display_label"
1864                    else {"node_label": anno_k}
1865                )
1866                node_validation_rules = dmge.get_node_validation_rules(**param)
1867
1868                if rule_in_rule_list("list", node_validation_rules):
1869                    annos["annotations"]["annotations"][anno_k] = anno_v.split(",")
1870                    continue
1871            # default: assign the original value
1872            annos["annotations"]["annotations"][anno_k] = anno_v
1873
1874        return annos
1875
1876    @async_missing_entity_handler
1877    async def format_row_annotations(
1878        self,
1879        dmge: DataModelGraphExplorer,
1880        row: pd.Series,
1881        entityId: str,
1882        hideBlanks: bool,
1883        annotation_keys: str,
1884    ) -> Union[None, Dict[str, Any]]:
1885        """Format row annotations
1886
1887        Args:
1888            dmge (DataModelGraphExplorer): data moodel graph explorer object
1889            row (pd.Series): row of the manifest
1890            entityId (str): entity id of the manifest
1891            hideBlanks (bool): when true, does not upload annotation keys with blank values. When false, upload Annotation keys with empty string values
1892            annotation_keys (str): display_label/class_label
1893
1894        Returns:
1895            Union[None, Dict[str,]]: if entity id is in trash can, return None. Otherwise, return the annotations
1896        """
1897        # prepare metadata for Synapse storage (resolve display name into a name that Synapse annotations support (e.g no spaces, parenthesis)
1898        # note: the removal of special characters, will apply only to annotation keys; we are not altering the manifest
1899        # this could create a divergence between manifest column and annotations. this should be ok for most use cases.
1900        # columns with special characters are outside of the schema
1901        metadataSyn = {}
1902        blacklist_chars = ["(", ")", ".", " ", "-"]
1903
1904        for k, v in row.to_dict().items():
1905            if annotation_keys == "display_label":
1906                keySyn = str(k).translate({ord(x): "" for x in blacklist_chars})
1907            elif annotation_keys == "class_label":
1908                keySyn = get_class_label_from_display_name(str(k)).translate(
1909                    {ord(x): "" for x in blacklist_chars}
1910                )
1911
1912            # Skip `Filename` and `ETag` columns when setting annotations
1913            if keySyn in ["Filename", "ETag", "eTag"]:
1914                continue
1915
1916            # truncate annotation values to 500 characters if the
1917            # size of values is greater than equal to 500 characters
1918            # add an explicit [truncatedByDataCuratorApp] message at the end
1919            # of every truncated message to indicate that the cell value
1920            # has been truncated
1921            if isinstance(v, str) and len(v) >= 500:
1922                v = v[0:472] + "[truncatedByDataCuratorApp]"
1923
1924            metadataSyn[keySyn] = v
1925
1926        # This will first check if the entity is already in memory, and if so, that
1927        # instance is used. Unfortunately, the expected return format needs to match
1928        # the Synapse API, so we need to convert the annotations to the expected format.
1929        entity = self.synapse_entity_tracker.get(
1930            synapse_id=entityId,
1931            syn=self.syn,
1932            download_file=False,
1933            retrieve_if_not_present=False,
1934        )
1935        if entity is not None:
1936            synapse_annotations = _convert_to_annotations_list(
1937                annotations=entity.annotations
1938            )
1939            annos = {
1940                "annotations": {
1941                    "id": entity.id,
1942                    "etag": entity.etag,
1943                    "annotations": synapse_annotations,
1944                }
1945            }
1946        else:
1947            annos = await self.get_async_annotation(entityId)
1948
1949        # set annotation(s) for the various objects/items in a dataset on Synapse
1950        csv_list_regex = comma_separated_list_regex()
1951
1952        annos = self.process_row_annotations(
1953            dmge=dmge,
1954            metadata_syn=metadataSyn,
1955            hide_blanks=hideBlanks,
1956            csv_list_regex=csv_list_regex,
1957            annos=annos,
1958            annotation_keys=annotation_keys,
1959        )
1960
1961        return annos
1962
1963    @missing_entity_handler
1964    @tracer.start_as_current_span("SynapseStorage::format_manifest_annotations")
1965    def format_manifest_annotations(self, manifest, manifest_synapse_id):
1966        """
1967        Set annotations for the manifest (as a whole) so they can be applied to the manifest table or csv.
1968        For now just getting the Component.
1969        """
1970
1971        entity = self.synapse_entity_tracker.get(
1972            synapse_id=manifest_synapse_id, syn=self.syn, download_file=False
1973        )
1974        is_file = entity.concreteType.endswith(".FileEntity")
1975        is_table = entity.concreteType.endswith(".TableEntity")
1976
1977        if is_file:
1978            # Get file metadata
1979            metadata = self.getFileAnnotations(manifest_synapse_id)
1980
1981            # If there is a defined component add it to the metadata.
1982            if "Component" in manifest.columns:
1983                # Gather component information
1984                component = manifest["Component"].unique()
1985
1986                # Double check that only a single component is listed, else raise an error.
1987                try:
1988                    len(component) == 1
1989                except ValueError as err:
1990                    raise ValueError(
1991                        f"Manifest has more than one component. Please check manifest and resubmit."
1992                    ) from err
1993
1994                # Add component to metadata
1995                metadata["Component"] = component[0]
1996
1997        elif is_table:
1998            # Get table metadata
1999            metadata = self.getTableAnnotations(manifest_synapse_id)
2000
2001        # Get annotations
2002        annos = OldAnnotations(
2003            id=entity.id, etag=entity.etag, values=entity.annotations
2004        )
2005
2006        # Add metadata to the annotations
2007        for annos_k, annos_v in metadata.items():
2008            annos[annos_k] = annos_v
2009
2010        return annos
2011
2012    '''
2013    def annotate_upload_manifest_table(self, manifest, datasetId, metadataManifestPath,
2014        useSchemaLabel: bool = True, hideBlanks: bool = False, restrict_manifest = False):
2015        """
2016        Purpose:
2017            Works very similarly to associateMetadataWithFiles except takes in the manifest
2018            rather than the manifest path
2019
2020        """
2021
2022        # Add uuid for table updates and fill.
2023        if not "Uuid" in manifest.columns:
2024            manifest["Uuid"] = ''
2025
2026        for idx,row in manifest.iterrows():
2027            if not row["Uuid"]:
2028                gen_uuid = uuid.uuid4()
2029                row["Uuid"] = gen_uuid
2030                manifest.loc[idx, 'Uuid'] = gen_uuid
2031
2032        # add entityId as a column if not already there or
2033        # fill any blanks with an empty string.
2034        if not "entityId" in manifest.columns:
2035            manifest["entityId"] = ""
2036        else:
2037            manifest["entityId"].fillna("", inplace=True)
2038
2039        # get a DataModelGraphExplorer object to ensure schema attribute names used in manifest are translated to schema labels for synapse annotations
2040        dmge = DataModelGraphExplorer()
2041
2042        # Create table name here.
2043        if 'Component' in manifest.columns:
2044            table_name = manifest['Component'][0].lower() + '_synapse_storage_manifest_table'
2045        else:
2046            table_name = 'synapse_storage_manifest_table'
2047
2048        # Upload manifest as a table and get the SynID and manifest
2049        manifest_synapse_table_id, manifest, table_manifest = self.upload_format_manifest_table(
2050                                                    dmge, manifest, datasetId, table_name, restrict = restrict_manifest, useSchemaLabel=useSchemaLabel,)
2051
2052        # Iterate over manifest rows, create Synapse entities and store corresponding entity IDs in manifest if needed
2053        # also set metadata for each synapse entity as Synapse annotations
2054        for idx, row in manifest.iterrows():
2055            if not row["entityId"]:
2056                # If not using entityIds, fill with manifest_table_id so
2057                row["entityId"] = manifest_synapse_table_id
2058                entityId = ''
2059            else:
2060                # get the entity id corresponding to this row
2061                entityId = row["entityId"]
2062
2063        # Load manifest to synapse as a CSV File
2064        manifest_synapse_file_id = self.upload_manifest_file(manifest, metadataManifestPath, datasetId, restrict_manifest)
2065
2066        # Get annotations for the file manifest.
2067        manifest_annotations = self.format_manifest_annotations(manifest, manifest_synapse_file_id)
2068
2069        self.syn.set_annotations(manifest_annotations)
2070
2071        logger.info("Associated manifest file with dataset on Synapse.")
2072
2073        # Update manifest Synapse table with new entity id column.
2074        self.make_synapse_table(
2075            table_to_load = table_manifest,
2076            dataset_id = datasetId,
2077            existingTableId = manifest_synapse_table_id,
2078            table_name = table_name,
2079            update_col = 'Uuid',
2080            specify_schema = False,
2081            )
2082
2083        # Get annotations for the table manifest
2084        manifest_annotations = self.format_manifest_annotations(manifest, manifest_synapse_table_id)
2085        self.syn.set_annotations(manifest_annotations)
2086        return manifest_synapse_table_id
2087    '''
2088
2089    def _read_manifest(self, metadataManifestPath: str) -> pd.DataFrame:
2090        """Helper function to read in provided manifest as a pandas DataFrame for subsequent downstream processing.
2091        Args:
2092            metadataManifestPath (str): path where manifest is stored
2093        Returns:
2094            manifest(pd.DataFrame): Manifest loaded as a pandas dataframe
2095        Raises:
2096            FileNotFoundError: Manifest file does not exist at provided path.
2097        """
2098        # read new manifest csv
2099        try:
2100            load_args = {
2101                "dtype": "string",
2102            }
2103            manifest = load_df(
2104                metadataManifestPath,
2105                preserve_raw_input=False,
2106                allow_na_values=False,
2107                **load_args,
2108            )
2109        except FileNotFoundError as err:
2110            raise FileNotFoundError(
2111                f"No manifest file was found at this path: {metadataManifestPath}"
2112            ) from err
2113        return manifest
2114
2115    def _add_id_columns_to_manifest(
2116        self, manifest: pd.DataFrame, dmge: DataModelGraphExplorer
2117    ):
2118        """Helper function to add id and entityId columns to the manifest if they do not already exist, Fill id values per row.
2119        Args:
2120            Manifest loaded as a pd.Dataframe
2121        Returns (pd.DataFrame):
2122            Manifest df with new Id and EntityId columns (and UUID values) if they were not already present.
2123        """
2124
2125        # Add Id for table updates and fill.
2126        if not col_in_dataframe("Id", manifest):
2127            # See if schema has `Uuid` column specified
2128            try:
2129                uuid_col_in_schema = dmge.is_class_in_schema(
2130                    "Uuid"
2131                ) or dmge.is_class_in_schema("uuid")
2132            except KeyError:
2133                uuid_col_in_schema = False
2134
2135            # Rename `Uuid` column if it wasn't specified in the schema
2136            if col_in_dataframe("Uuid", manifest) and not uuid_col_in_schema:
2137                manifest.rename(columns={"Uuid": "Id"}, inplace=True)
2138            # If no `Uuid` column exists or it is specified in the schema, create a new `Id` column
2139            else:
2140                manifest["Id"] = ""
2141
2142        # Retrieve the ID column name (id, Id and ID) are treated the same.
2143        id_col_name = [col for col in manifest.columns if col.lower() == "id"][0]
2144
2145        # Check if values have been added to the Id coulumn, if not add a UUID so value in the row is not blank.
2146        for idx, row in manifest.iterrows():
2147            if not row[id_col_name]:
2148                gen_uuid = str(uuid.uuid4())
2149                row[id_col_name] = gen_uuid
2150                manifest.loc[idx, id_col_name] = gen_uuid
2151
2152        # add entityId as a column if not already there or
2153        # fill any blanks with an empty string.
2154        if not col_in_dataframe("entityId", manifest):
2155            manifest["entityId"] = ""
2156        else:
2157            manifest["entityId"].fillna("", inplace=True)
2158
2159        return manifest
2160
2161    def _generate_table_name(self, manifest):
2162        """Helper function to generate a table name for upload to synapse.
2163
2164        Args:
2165            Manifest loaded as a pd.Dataframe
2166
2167        Returns:
2168            table_name (str): Name of the table to load
2169            component_name (str): Name of the manifest component (if applicable)
2170        """
2171        # Create table name here.
2172        if "Component" in manifest.columns:
2173            component_name = manifest["Component"][0].lower()
2174            table_name = component_name + "_synapse_storage_manifest_table"
2175        else:
2176            component_name = ""
2177            table_name = "synapse_storage_manifest_table"
2178        return table_name, component_name
2179
2180    def _create_entity_id(self, idx, row, manifest, datasetId):
2181        """Helper function to generate an entityId and add it to the appropriate row in the manifest.
2182        Args:
2183            row: current row of manifest being processed
2184            manifest (pd.DataFrame): loaded df containing user supplied data.
2185            datasetId (str): synapse ID of folder containing the dataset
2186
2187        Returns:
2188            manifest (pd.DataFrame): manifest with entityId added to the appropriate row
2189            entityId (str): Generated Entity Id.
2190
2191        """
2192        rowEntity = Folder(str(uuid.uuid4()), parent=datasetId)
2193        rowEntity = self.syn.store(rowEntity)
2194        entityId = rowEntity["id"]
2195        self.synapse_entity_tracker.add(synapse_id=entityId, entity=rowEntity)
2196        row["entityId"] = entityId
2197        manifest.loc[idx, "entityId"] = entityId
2198        return manifest, entityId
2199
2200    async def _process_store_annos(self, requests: Set[asyncio.Task]) -> None:
2201        """Process annotations and store them on synapse asynchronously
2202
2203        Args:
2204            requests (Set[asyncio.Task]): a set of tasks of formatting annotations created by format_row_annotations function in previous step
2205
2206        Raises:
2207            RuntimeError: raise a run time error if a task failed to complete
2208        """
2209        while requests:
2210            done_tasks, pending_tasks = await asyncio.wait(
2211                requests, return_when=asyncio.FIRST_COMPLETED
2212            )
2213            requests = pending_tasks
2214
2215            for completed_task in done_tasks:
2216                try:
2217                    annos = completed_task.result()
2218
2219                    if isinstance(annos, Annotations):
2220                        logger.info(f"Successfully stored annotations for {annos.id}")
2221                    else:
2222                        # store annotations if they are not None
2223                        if annos:
2224                            entity_id = annos["annotations"]["id"]
2225                            logger.info(
2226                                f"Obtained and processed annotations for {entity_id} entity"
2227                            )
2228                            requests.add(
2229                                asyncio.create_task(
2230                                    self.store_async_annotation(annotation_dict=annos)
2231                                )
2232                            )
2233                except Exception as e:
2234                    raise RuntimeError(f"failed with { repr(e) }.") from e
2235
2236    @tracer.start_as_current_span("SynapseStorage::add_annotations_to_entities_files")
2237    async def add_annotations_to_entities_files(
2238        self,
2239        dmge,
2240        manifest,
2241        manifest_record_type: str,
2242        datasetId: str,
2243        hideBlanks: bool,
2244        manifest_synapse_table_id="",
2245        annotation_keys: str = "class_label",
2246    ):
2247        """
2248        Depending on upload type add Ids to entityId row. Add anotations to connected
2249        files and folders. Despite the name of this function, it also applies to folders.
2250
2251        Args:
2252            dmge: DataModelGraphExplorer Object
2253            manifest (pd.DataFrame): loaded df containing user supplied data.
2254            manifest_record_type: valid values are 'entity', 'table' or 'both'. Specifies whether to create entity ids and folders for each row in a manifest, a Synapse table to house the entire manifest or do both.
2255            datasetId (str): synapse ID of folder containing the dataset
2256            hideBlanks (bool): Default is false -Boolean flag that does not upload annotation keys with blank values when true. Uploads Annotation keys with empty string values when false.
2257            manifest_synapse_table_id (str): Default is an empty string ''.
2258            annotation_keys: (str) display_label/class_label(default), Determines labeling syle for annotation keys. class_label will format the display
2259                name as upper camelcase, and strip blacklisted characters, display_label will strip blacklisted characters including spaces, to retain
2260                display label formatting while ensuring the label is formatted properly for Synapse annotations.
2261        Returns:
2262            manifest (pd.DataFrame): modified to add entitiyId as appropriate
2263
2264        """
2265
2266        # Expected behavior is to annotate files if `Filename` is present and if file_annotations_upload is set to True regardless of `-mrt` setting
2267        if "filename" in [col.lower() for col in manifest.columns]:
2268            # get current list of files and store as dataframe
2269            dataset_files = self.getFilesInStorageDataset(datasetId)
2270            files_and_entityIds = self._get_file_entityIds(
2271                dataset_files=dataset_files, only_new_files=False
2272            )
2273            file_df = pd.DataFrame(files_and_entityIds)
2274
2275            # Merge dataframes to add entityIds
2276            manifest = manifest.merge(
2277                file_df, how="left", on="Filename", suffixes=["_x", None]
2278            ).drop("entityId_x", axis=1)
2279
2280        # Fill `entityId` for each row if missing and annotate entity as appropriate
2281        requests = set()
2282        for idx, row in manifest.iterrows():
2283            if not row["entityId"] and (
2284                manifest_record_type == "file_and_entities"
2285                or manifest_record_type == "table_file_and_entities"
2286            ):
2287                manifest, entityId = self._create_entity_id(
2288                    idx, row, manifest, datasetId
2289                )
2290            elif not row["entityId"] and manifest_record_type == "table_and_file":
2291                # If not using entityIds, fill with manifest_table_id so
2292                row["entityId"] = manifest_synapse_table_id
2293                manifest.loc[idx, "entityId"] = manifest_synapse_table_id
2294                entityId = ""
2295                # If the row is the manifest table, do not add annotations
2296            elif row["entityId"] == manifest_synapse_table_id:
2297                entityId = ""
2298            else:
2299                # get the file id of the file to annotate, collected in above step.
2300                entityId = row["entityId"]
2301
2302            # Adding annotations to connected files.
2303            if entityId:
2304                # Format annotations for Synapse
2305                annos_task = asyncio.create_task(
2306                    self.format_row_annotations(
2307                        dmge, row, entityId, hideBlanks, annotation_keys
2308                    )
2309                )
2310                requests.add(annos_task)
2311        await self._process_store_annos(requests)
2312        return manifest
2313
2314    @tracer.start_as_current_span("SynapseStorage::upload_manifest_as_table")
2315    def upload_manifest_as_table(
2316        self,
2317        dmge: DataModelGraphExplorer,
2318        manifest: pd.DataFrame,
2319        metadataManifestPath: str,
2320        datasetId: str,
2321        table_name: str,
2322        component_name: str,
2323        restrict: bool,
2324        manifest_record_type: str,
2325        hideBlanks: bool,
2326        table_manipulation: str,
2327        table_column_names: str,
2328        annotation_keys: str,
2329        file_annotations_upload: bool = True,
2330    ):
2331        """Upload manifest to Synapse as a table and csv.
2332        Args:
2333            dmge: DataModelGraphExplorer object
2334            manifest (pd.DataFrame): loaded df containing user supplied data.
2335            metadataManifestPath: path to csv containing a validated metadata manifest.
2336            datasetId (str): synapse ID of folder containing the dataset
2337            table_name (str): Generated to name the table being uploaded.
2338            component_name (str): Name of the component manifest that is currently being uploaded.
2339            restrict (bool): Flag for censored data.
2340            manifest_record_type (str): valid values are 'entity', 'table' or 'both'. Specifies whether to create entity ids and folders for each row in a manifest, a Synapse table to house the entire manifest or do both.
2341            hideBlanks (bool): Default is False -Boolean flag that does not upload annotation keys with blank values when true. Uploads Annotation keys with empty string values when false.
2342            table_malnipulation (str): Specify the way the manifest tables should be store as on Synapse when one with the same name already exists. Options are 'replace' and 'upsert'.
2343            table_column_names: (str): display_name/display_label/class_label (default). Sets labeling style for table column names. display_name will use the raw display name as the column name. class_label will format the display
2344                name as upper camelcase, and strip blacklisted characters, display_label will strip blacklisted characters including spaces, to retain
2345                display label formatting.
2346            annotation_keys: (str) display_label/class_label (default), Sets labeling syle for annotation keys. class_label will format the display
2347                name as upper camelcase, and strip blacklisted characters, display_label will strip blacklisted characters including spaces, to retain
2348                display label formatting while ensuring the label is formatted properly for Synapse annotations.
2349            file_annotations_upload (bool): Default to True. If false, do not add annotations to files.
2350        Return:
2351            manifest_synapse_file_id: SynID of manifest csv uploaded to synapse.
2352        """
2353        # Upload manifest as a table, get the ID and updated manifest.
2354        manifest_synapse_table_id, manifest, table_manifest = self.uploadDB(
2355            dmge=dmge,
2356            manifest=manifest,
2357            datasetId=datasetId,
2358            table_name=table_name,
2359            restrict=restrict,
2360            table_manipulation=table_manipulation,
2361            table_column_names=table_column_names,
2362        )
2363
2364        if file_annotations_upload:
2365            manifest = asyncio.run(
2366                self.add_annotations_to_entities_files(
2367                    dmge,
2368                    manifest,
2369                    manifest_record_type,
2370                    datasetId,
2371                    hideBlanks,
2372                    manifest_synapse_table_id,
2373                    annotation_keys,
2374                )
2375            )
2376        # Load manifest to synapse as a CSV File
2377        manifest_synapse_file_id = self.upload_manifest_file(
2378            manifest=manifest,
2379            metadataManifestPath=metadataManifestPath,
2380            datasetId=datasetId,
2381            restrict_manifest=restrict,
2382            component_name=component_name,
2383        )
2384
2385        # Set annotations for the file manifest.
2386        manifest_annotations = self.format_manifest_annotations(
2387            manifest=manifest, manifest_synapse_id=manifest_synapse_file_id
2388        )
2389        annos = self.syn.set_annotations(annotations=manifest_annotations)
2390        manifest_entity = self.synapse_entity_tracker.get(
2391            synapse_id=manifest_synapse_file_id, syn=self.syn, download_file=False
2392        )
2393        manifest_entity.annotations = annos
2394        manifest_entity.etag = annos.etag
2395
2396        logger.info("Associated manifest file with dataset on Synapse.")
2397
2398        # Update manifest Synapse table with new entity id column.
2399        manifest_synapse_table_id, manifest, _ = self.uploadDB(
2400            dmge=dmge,
2401            manifest=manifest,
2402            datasetId=datasetId,
2403            table_name=table_name,
2404            restrict=restrict,
2405            table_manipulation="update",
2406            table_column_names=table_column_names,
2407        )
2408
2409        # Set annotations for the table manifest
2410        manifest_annotations = self.format_manifest_annotations(
2411            manifest=manifest, manifest_synapse_id=manifest_synapse_table_id
2412        )
2413        annotations_manifest_table = self.syn.set_annotations(
2414            annotations=manifest_annotations
2415        )
2416        manifest_table_entity = self.synapse_entity_tracker.get(
2417            synapse_id=manifest_synapse_table_id, syn=self.syn, download_file=False
2418        )
2419        manifest_table_entity.annotations = annotations_manifest_table
2420        manifest_table_entity.etag = annotations_manifest_table.etag
2421
2422        return manifest_synapse_file_id
2423
2424    @tracer.start_as_current_span("SynapseStorage::upload_manifest_as_csv")
2425    def upload_manifest_as_csv(
2426        self,
2427        dmge,
2428        manifest,
2429        metadataManifestPath,
2430        datasetId,
2431        restrict,
2432        manifest_record_type,
2433        hideBlanks,
2434        component_name,
2435        annotation_keys: str,
2436        file_annotations_upload: bool = True,
2437    ):
2438        """Upload manifest to Synapse as a csv only.
2439        Args:
2440            dmge: DataModelGraphExplorer object
2441            manifest (pd.DataFrame): loaded df containing user supplied data.
2442            metadataManifestPath: path to csv containing a validated metadata manifest.
2443            datasetId (str): synapse ID of folder containing the dataset
2444            restrict (bool): Flag for censored data.
2445            manifest_record_type: valid values are 'entity', 'table' or 'both'. Specifies whether to create entity ids and folders for each row in a manifest, a Synapse table to house the entire manifest or do both.
2446            hideBlanks (bool): Default is False -Boolean flag that does not upload annotation keys with blank values when true. Uploads Annotation keys with empty string values when false.
2447            annotation_keys: (str) display_label/class_label (default), Sets labeling syle for annotation keys. class_label will format the display
2448                name as upper camelcase, and strip blacklisted characters, display_label will strip blacklisted characters including spaces, to retain
2449                display label formatting while ensuring the label is formatted properly for Synapse annotations.
2450            file_annotations_upload (bool): Default to True. If false, do not add annotations to files.
2451        Return:
2452            manifest_synapse_file_id (str): SynID of manifest csv uploaded to synapse.
2453        """
2454        if file_annotations_upload:
2455            manifest = asyncio.run(
2456                self.add_annotations_to_entities_files(
2457                    dmge,
2458                    manifest,
2459                    manifest_record_type,
2460                    datasetId,
2461                    hideBlanks,
2462                    annotation_keys=annotation_keys,
2463                )
2464            )
2465
2466        # Load manifest to synapse as a CSV File
2467        manifest_synapse_file_id = self.upload_manifest_file(
2468            manifest,
2469            metadataManifestPath,
2470            datasetId,
2471            restrict,
2472            component_name=component_name,
2473        )
2474
2475        # Set annotations for the file manifest.
2476        manifest_annotations = self.format_manifest_annotations(
2477            manifest, manifest_synapse_file_id
2478        )
2479        annos = self.syn.set_annotations(manifest_annotations)
2480        manifest_entity = self.synapse_entity_tracker.get(
2481            synapse_id=manifest_synapse_file_id, syn=self.syn, download_file=False
2482        )
2483        manifest_entity.annotations = annos
2484        manifest_entity.etag = annos.etag
2485
2486        logger.info("Associated manifest file with dataset on Synapse.")
2487
2488        return manifest_synapse_file_id
2489
2490    @tracer.start_as_current_span("SynapseStorage::upload_manifest_combo")
2491    def upload_manifest_combo(
2492        self,
2493        dmge,
2494        manifest,
2495        metadataManifestPath,
2496        datasetId,
2497        table_name,
2498        component_name,
2499        restrict,
2500        manifest_record_type,
2501        hideBlanks,
2502        table_manipulation,
2503        table_column_names: str,
2504        annotation_keys: str,
2505        file_annotations_upload: bool = True,
2506    ):
2507        """Upload manifest to Synapse as a table and CSV with entities.
2508        Args:
2509            dmge: DataModelGraphExplorer object
2510            manifest (pd.DataFrame): loaded df containing user supplied data.
2511            metadataManifestPath: path to csv containing a validated metadata manifest.
2512            datasetId (str): synapse ID of folder containing the dataset
2513            table_name (str): Generated to name the table being uploaded.
2514            component_name (str): Name of the component manifest that is currently being uploaded.
2515            restrict (bool): Flag for censored data.
2516            manifest_record_type: valid values are 'entity', 'table' or 'both'. Specifies whether to create entity ids and folders for each row in a manifest, a Synapse table to house the entire manifest or do both.
2517            hideBlanks (bool): Default is False -Boolean flag that does not upload annotation keys with blank values when true. Uploads Annotation keys with empty string values when false.
2518            table_malnipulation (str): Specify the way the manifest tables should be store as on Synapse when one with the same name already exists. Options are 'replace' and 'upsert'.
2519            table_column_names: (str): display_name/display_label/class_label (default). Sets labeling style for table column names. display_name will use the raw display name as the column name. class_label will format the display
2520                name as upper camelcase, and strip blacklisted characters, display_label will strip blacklisted characters including spaces, to retain
2521                display label formatting.
2522            annotation_keys: (str) display_label/class_label (default), Sets labeling syle for annotation keys. class_label will format the display
2523                name as upper camelcase, and strip blacklisted characters, display_label will strip blacklisted characters including spaces, to retain
2524                display label formatting while ensuring the label is formatted properly for Synapse annotations.
2525            file_annotations_upload (bool): Default to True. If false, do not add annotations to files.
2526        Return:
2527            manifest_synapse_file_id (str): SynID of manifest csv uploaded to synapse.
2528        """
2529        manifest_synapse_table_id, manifest, table_manifest = self.uploadDB(
2530            dmge=dmge,
2531            manifest=manifest,
2532            datasetId=datasetId,
2533            table_name=table_name,
2534            restrict=restrict,
2535            table_manipulation=table_manipulation,
2536            table_column_names=table_column_names,
2537        )
2538
2539        if file_annotations_upload:
2540            manifest = asyncio.run(
2541                self.add_annotations_to_entities_files(
2542                    dmge,
2543                    manifest,
2544                    manifest_record_type,
2545                    datasetId,
2546                    hideBlanks,
2547                    manifest_synapse_table_id,
2548                    annotation_keys=annotation_keys,
2549                )
2550            )
2551
2552        # Load manifest to synapse as a CSV File
2553        manifest_synapse_file_id = self.upload_manifest_file(
2554            manifest, metadataManifestPath, datasetId, restrict, component_name
2555        )
2556
2557        # Set annotations for the file manifest.
2558        manifest_annotations = self.format_manifest_annotations(
2559            manifest, manifest_synapse_file_id
2560        )
2561        file_manifest_annoations = self.syn.set_annotations(manifest_annotations)
2562        manifest_entity = self.synapse_entity_tracker.get(
2563            synapse_id=manifest_synapse_file_id, syn=self.syn, download_file=False
2564        )
2565        manifest_entity.annotations = file_manifest_annoations
2566        manifest_entity.etag = file_manifest_annoations.etag
2567        logger.info("Associated manifest file with dataset on Synapse.")
2568
2569        # Update manifest Synapse table with new entity id column.
2570        manifest_synapse_table_id, manifest, table_manifest = self.uploadDB(
2571            dmge=dmge,
2572            manifest=manifest,
2573            datasetId=datasetId,
2574            table_name=table_name,
2575            restrict=restrict,
2576            table_manipulation="update",
2577            table_column_names=table_column_names,
2578        )
2579
2580        # Set annotations for the table manifest
2581        manifest_annotations = self.format_manifest_annotations(
2582            manifest, manifest_synapse_table_id
2583        )
2584        table_manifest_annotations = self.syn.set_annotations(manifest_annotations)
2585        manifest_entity = self.synapse_entity_tracker.get(
2586            synapse_id=manifest_synapse_table_id, syn=self.syn, download_file=False
2587        )
2588        manifest_entity.annotations = table_manifest_annotations
2589        manifest_entity.etag = table_manifest_annotations.etag
2590        return manifest_synapse_file_id
2591
2592    @tracer.start_as_current_span("SynapseStorage::associateMetadataWithFiles")
2593    def associateMetadataWithFiles(
2594        self,
2595        dmge: DataModelGraphExplorer,
2596        metadataManifestPath: str,
2597        datasetId: str,
2598        manifest_record_type: str = "table_file_and_entities",
2599        hideBlanks: bool = False,
2600        restrict_manifest=False,
2601        table_manipulation: str = "replace",
2602        table_column_names: str = "class_label",
2603        annotation_keys: str = "class_label",
2604        file_annotations_upload: bool = True,
2605    ) -> str:
2606        """Associate metadata with files in a storage dataset already on Synapse.
2607        Upload metadataManifest in the storage dataset folder on Synapse as well. Return synapseId of the uploaded manifest file.
2608
2609        If this is a new manifest there could be no Synapse entities associated with the rows of this manifest
2610        this may be due to data type (e.g. clinical data) being tabular
2611        and not requiring files; to utilize uniform interfaces downstream
2612        (i.e. fileviews), a Synapse entity (a folder) is created for each row
2613        and an entity column is added to the manifest containing the resulting
2614        entity IDs; a table is also created at present as an additional interface
2615        for downstream query and interaction with the data.
2616
2617        Args:
2618            dmge: DataModelGraphExplorer Object
2619            metadataManifestPath: path to csv containing a validated metadata manifest.
2620            The manifest should include a column entityId containing synapse IDs of files/entities to be associated with metadata, if that is applicable to the dataset type.
2621            Some datasets, e.g. clinical data, do not contain file id's, but data is stored in a table: one row per item.
2622            In this case, the system creates a file on Synapse for each row in the table (e.g. patient, biospecimen) and associates the columnset data as metadata/annotations to his file.
2623            datasetId: synapse ID of folder containing the dataset
2624            manifest_record_type: Default value is 'table_file_and_entities'. valid values are 'file_only', 'file_and_entities', 'table_and_file' or 'table_file_and_entities'. 'file_and_entities' will store the manifest as a csv and create Synapse files for each row in the manifest.'table_and_file' will store the manifest as a table and a csv on Synapse. 'file_only' will store the manifest as a csv only on Synapse. 'table_file_and_entities' will perform the options file_with_entites and table in combination.
2625            hideBlanks: Default is false. Boolean flag that does not upload annotation keys with blank values when true. Uploads Annotation keys with empty string values when false.
2626            restrict_manifest (bool): Default is false. Flag for censored data.
2627            table_malnipulation (str): Default is 'replace'. Specify the way the manifest tables should be store as on Synapse when one with the same name already exists. Options are 'replace' and 'upsert'.
2628            table_column_names: (str): display_name/display_label/class_label (default). Sets labeling style for table column names. display_name will use the raw display name as the column name. class_label will format the display
2629                name as upper camelcase, and strip blacklisted characters, display_label will strip blacklisted characters including spaces, to retain
2630                display label formatting.
2631            annotation_keys: (str) display_label/class_label (default), Sets labeling syle for annotation keys. class_label will format the display
2632                name as upper camelcase, and strip blacklisted characters, display_label will strip blacklisted characters including spaces, to retain
2633                display label formatting while ensuring the label is formatted properly for Synapse annotations.
2634        Returns:
2635            manifest_synapse_file_id: SynID of manifest csv uploaded to synapse.
2636        """
2637        # Read new manifest CSV:
2638        manifest = self._read_manifest(metadataManifestPath)
2639        manifest = self._add_id_columns_to_manifest(manifest, dmge)
2640
2641        table_name, component_name = self._generate_table_name(manifest)
2642
2643        # Upload manifest to synapse based on user input (manifest_record_type)
2644        if manifest_record_type == "file_only":
2645            manifest_synapse_file_id = self.upload_manifest_as_csv(
2646                dmge=dmge,
2647                manifest=manifest,
2648                metadataManifestPath=metadataManifestPath,
2649                datasetId=datasetId,
2650                restrict=restrict_manifest,
2651                hideBlanks=hideBlanks,
2652                manifest_record_type=manifest_record_type,
2653                component_name=component_name,
2654                annotation_keys=annotation_keys,
2655                file_annotations_upload=file_annotations_upload,
2656            )
2657        elif manifest_record_type == "table_and_file":
2658            manifest_synapse_file_id = self.upload_manifest_as_table(
2659                dmge=dmge,
2660                manifest=manifest,
2661                metadataManifestPath=metadataManifestPath,
2662                datasetId=datasetId,
2663                table_name=table_name,
2664                component_name=component_name,
2665                restrict=restrict_manifest,
2666                hideBlanks=hideBlanks,
2667                manifest_record_type=manifest_record_type,
2668                table_manipulation=table_manipulation,
2669                table_column_names=table_column_names,
2670                annotation_keys=annotation_keys,
2671                file_annotations_upload=file_annotations_upload,
2672            )
2673        elif manifest_record_type == "file_and_entities":
2674            manifest_synapse_file_id = self.upload_manifest_as_csv(
2675                dmge=dmge,
2676                manifest=manifest,
2677                metadataManifestPath=metadataManifestPath,
2678                datasetId=datasetId,
2679                restrict=restrict_manifest,
2680                hideBlanks=hideBlanks,
2681                manifest_record_type=manifest_record_type,
2682                component_name=component_name,
2683                annotation_keys=annotation_keys,
2684                file_annotations_upload=file_annotations_upload,
2685            )
2686        elif manifest_record_type == "table_file_and_entities":
2687            manifest_synapse_file_id = self.upload_manifest_combo(
2688                dmge=dmge,
2689                manifest=manifest,
2690                metadataManifestPath=metadataManifestPath,
2691                datasetId=datasetId,
2692                table_name=table_name,
2693                component_name=component_name,
2694                restrict=restrict_manifest,
2695                hideBlanks=hideBlanks,
2696                manifest_record_type=manifest_record_type,
2697                table_manipulation=table_manipulation,
2698                table_column_names=table_column_names,
2699                annotation_keys=annotation_keys,
2700                file_annotations_upload=file_annotations_upload,
2701            )
2702        else:
2703            raise ValueError("Please enter a valid manifest_record_type.")
2704        return manifest_synapse_file_id
2705
2706    def getTableAnnotations(self, table_id: str):
2707        """Generate dictionary of annotations for the given Synapse file.
2708        Synapse returns all custom annotations as lists since they
2709        can contain multiple values. In all cases, the values will
2710        be converted into strings and concatenated with ", ".
2711
2712        Args:
2713            fileId (str): Synapse ID for dataset file.
2714
2715        Returns:
2716            dict: Annotations as comma-separated strings.
2717        """
2718        try:
2719            entity = self.synapse_entity_tracker.get(
2720                synapse_id=table_id, syn=self.syn, download_file=False
2721            )
2722            is_table = entity.concreteType.endswith(".TableEntity")
2723            annotations_raw = entity.annotations
2724        except SynapseHTTPError:
2725            # If an error occurs with retrieving entity, skip it
2726            # This could be caused by a temporary file view that
2727            # was deleted since its ID was retrieved
2728            is_file, is_table = False, False
2729
2730        # Skip anything that isn't a file or folder
2731        if not (is_table):
2732            return None
2733
2734        annotations = self.getEntityAnnotations(table_id, entity, annotations_raw)
2735
2736        return annotations
2737
2738    def getFileAnnotations(self, fileId: str) -> Dict[str, str]:
2739        """Generate dictionary of annotations for the given Synapse file.
2740        Synapse returns all custom annotations as lists since they
2741        can contain multiple values. In all cases, the values will
2742        be converted into strings and concatenated with ", ".
2743
2744        Args:
2745            fileId (str): Synapse ID for dataset file.
2746
2747        Returns:
2748            dict: Annotations as comma-separated strings.
2749        """
2750
2751        # Get entity metadata, including annotations
2752        try:
2753            entity = self.synapse_entity_tracker.get(
2754                synapse_id=fileId, syn=self.syn, download_file=False
2755            )
2756            is_file = entity.concreteType.endswith(".FileEntity")
2757            is_folder = entity.concreteType.endswith(".Folder")
2758            annotations_raw = entity.annotations
2759        except SynapseHTTPError:
2760            # If an error occurs with retrieving entity, skip it
2761            # This could be caused by a temporary file view that
2762            # was deleted since its ID was retrieved
2763            is_file, is_folder = False, False
2764
2765        # Skip anything that isn't a file or folder
2766        if not (is_file or is_folder):
2767            return None
2768
2769        annotations = self.getEntityAnnotations(fileId, entity, annotations_raw)
2770
2771        return annotations
2772
2773    def getEntityAnnotations(self, fileId, entity, annotations_raw):
2774        # Extract annotations from their lists and stringify. For example:
2775        # {'YearofBirth': [1980], 'author': ['bruno', 'milen', 'sujay']}
2776        annotations = dict()
2777        for key, vals in annotations_raw.items():
2778            if isinstance(vals, list) and len(vals) == 1:
2779                annotations[key] = str(vals[0])
2780            else:
2781                annotations[key] = ", ".join(str(v) for v in vals)
2782
2783        # Add the file entity ID and eTag, which weren't lists
2784        assert fileId == entity.id, (
2785            "For some reason, the Synapse ID in the response doesn't match"
2786            "the Synapse ID sent in the request (via synapseclient)."
2787        )
2788        annotations["entityId"] = fileId
2789        annotations["eTag"] = entity.etag
2790
2791        return annotations
2792
2793    def getDatasetAnnotations(
2794        self, datasetId: str, fill_na: bool = True, force_batch: bool = False
2795    ) -> pd.DataFrame:
2796        """Generate table for annotations across all files in given dataset.
2797
2798        Args:
2799            datasetId (str): Synapse ID for dataset folder.
2800            fill_na (bool): Whether to replace missing values with
2801                blank strings.
2802            force_batch (bool): Whether to force the function to use
2803                the batch mode, which uses a file view to retrieve
2804                annotations for a given dataset. Default to False
2805                unless there are more than 50 files in the dataset.
2806
2807        Returns:
2808            pd.DataFrame: Table of annotations.
2809        """
2810        # Get all files in given dataset
2811        dataset_files = self.getFilesInStorageDataset(datasetId)
2812
2813        # if there are no dataset files, there are no annotations
2814        # return None
2815        if not dataset_files:
2816            return pd.DataFrame()
2817
2818        dataset_files_map = dict(dataset_files)
2819        dataset_file_ids, _ = list(zip(*dataset_files))
2820
2821        # Get annotations for each file from Step 1
2822        # Batch mode
2823        try_batch = len(dataset_files) >= 50 or force_batch
2824        if try_batch:
2825            try:
2826                logger.info("Trying batch mode for retrieving Synapse annotations")
2827                table = self.getDatasetAnnotationsBatch(datasetId, dataset_file_ids)
2828            except (SynapseAuthenticationError, SynapseHTTPError, ValueError):
2829                logger.info(
2830                    f"Unable to create a temporary file view bound to {datasetId}. "
2831                    "Defaulting to slower iterative retrieval of annotations."
2832                )
2833                # Default to the slower non-batch method
2834                logger.info("Batch mode failed (probably due to permission error)")
2835                try_batch = False
2836
2837        # Non-batch mode
2838        if not try_batch:
2839            logger.info("Using slower (non-batch) sequential mode")
2840            records = [self.getFileAnnotations(i) for i in dataset_file_ids]
2841            # Remove any annotations for non-file/folders (stored as None)
2842            records = filter(None, records)
2843            table = pd.DataFrame.from_records(records)
2844
2845        # Add filenames for the files that "survived" annotation retrieval
2846        filenames = [dataset_files_map[i] for i in table["entityId"]]
2847
2848        if "Filename" not in table.columns:
2849            table.insert(0, "Filename", filenames)
2850
2851        # Ensure that entityId and eTag are at the end
2852        entity_ids = table.pop("entityId")
2853        etags = table.pop("eTag")
2854        table.insert(len(table.columns), "entityId", entity_ids)
2855        table.insert(len(table.columns), "eTag", etags)
2856
2857        # Missing values are filled in with empty strings for Google Sheets
2858        if fill_na:
2859            table.fillna("", inplace=True)
2860
2861        # Force all values as strings
2862        return table.astype(str)
2863
2864    def raise_final_error(retry_state):
2865        return retry_state.outcome.result()
2866
2867    def checkIfinAssetView(self, syn_id) -> str:
2868        # get data in administrative fileview for this pipeline
2869        assetViewTable = self.getStorageFileviewTable()
2870        all_files = list(assetViewTable["id"])
2871        if syn_id in all_files:
2872            return True
2873        else:
2874            return False
2875
2876    @tracer.start_as_current_span("SynapseStorage::getDatasetProject")
2877    @retry(
2878        stop=stop_after_attempt(5),
2879        wait=wait_chain(
2880            *[wait_fixed(10) for i in range(2)]
2881            + [wait_fixed(15) for i in range(2)]
2882            + [wait_fixed(20)]
2883        ),
2884        retry=retry_if_exception_type(LookupError),
2885        retry_error_callback=raise_final_error,
2886    )
2887    def getDatasetProject(self, datasetId: str) -> str:
2888        """Get parent project for a given dataset ID.
2889
2890        Args:
2891            datasetId (str): Synapse entity ID (folder or project).
2892
2893        Raises:
2894            ValueError: Raised if Synapse ID cannot be retrieved
2895            by the user or if it doesn't appear in the file view.
2896
2897        Returns:
2898            str: The Synapse ID for the parent project.
2899        """
2900
2901        # Subset main file view
2902        dataset_index = self.storageFileviewTable["id"] == datasetId
2903        dataset_row = self.storageFileviewTable[dataset_index]
2904
2905        # re-query if no datasets found
2906        if dataset_row.empty:
2907            sleep(5)
2908            self.query_fileview(force_requery=True)
2909            # Subset main file view
2910            dataset_index = self.storageFileviewTable["id"] == datasetId
2911            dataset_row = self.storageFileviewTable[dataset_index]
2912
2913        # Return `projectId` for given row if only one found
2914        if len(dataset_row) == 1:
2915            dataset_project = dataset_row["projectId"].values[0]
2916            return dataset_project
2917
2918        # Otherwise, check if already project itself
2919        try:
2920            syn_object = self.synapse_entity_tracker.get(
2921                synapse_id=datasetId, syn=self.syn, download_file=False
2922            )
2923            if syn_object.properties["concreteType"].endswith("Project"):
2924                return datasetId
2925        except SynapseHTTPError:
2926            raise PermissionError(
2927                f"The given dataset ({datasetId}) isn't accessible with this "
2928                "user. This might be caused by a typo in the dataset Synapse ID."
2929            )
2930
2931        # If not, then assume dataset not in file view
2932        raise LookupError(
2933            f"The given dataset ({datasetId}) doesn't appear in the "
2934            f"configured file view ({self.storageFileview}). This might "
2935            "mean that the file view's scope needs to be updated."
2936        )
2937
2938    def getDatasetAnnotationsBatch(
2939        self, datasetId: str, dataset_file_ids: Sequence[str] = None
2940    ) -> pd.DataFrame:
2941        """Generate table for annotations across all files in given dataset.
2942        This function uses a temporary file view to generate a table
2943        instead of iteratively querying for individual entity annotations.
2944        This function is expected to run much faster than
2945        `self.getDatasetAnnotationsBatch` on large datasets.
2946
2947        Args:
2948            datasetId (str): Synapse ID for dataset folder.
2949            dataset_file_ids (Sequence[str]): List of Synapse IDs
2950                for dataset files/folders used to subset the table.
2951
2952        Returns:
2953            pd.DataFrame: Table of annotations.
2954        """
2955        # Create data frame from annotations file view
2956        with DatasetFileView(datasetId, self.syn) as fileview:
2957            table = fileview.query()
2958
2959        if dataset_file_ids:
2960            table = table.loc[table.index.intersection(dataset_file_ids)]
2961
2962        table = table.reset_index(drop=True)
2963
2964        return table
2965
2966    def _get_table_schema_by_cname(self, table_schema):
2967        # assume no duplicate column names in the table
2968        table_schema_by_cname = {}
2969
2970        for col_record in table_schema:
2971            # TODO clean up dictionary for compactness (e.g. remove redundant 'name' key)
2972            table_schema_by_cname[col_record["name"]] = col_record
2973
2974        return table_schema_by_cname
2975
2976
2977class TableOperations:
2978    """
2979    Object to hold functions for various table operations specific to the Synapse Asset Store.
2980
2981    Currently implement operations are:
2982    createTable: upload a manifest as a new table when none exist
2983    replaceTable: replace a metadata in a table from one manifest with metadata from another manifest
2984    updateTable: add a column to a table that already exists on synapse
2985
2986    Operations currently in development are:
2987    upsertTable: add metadata from a manifest to an existing table that contains metadata from another manifest
2988    """
2989
2990    def __init__(
2991        self,
2992        synStore: SynapseStorage,
2993        tableToLoad: pd.DataFrame = None,
2994        tableName: str = None,
2995        datasetId: str = None,
2996        existingTableId: str = None,
2997        restrict: bool = False,
2998        synapse_entity_tracker: SynapseEntityTracker = None,
2999    ):
3000        """
3001        Class governing table operations (creation, replacement, upserts, updates) in schematic
3002
3003        tableToLoad: manifest formatted appropriately for the table
3004        tableName: name of the table to be uploaded
3005        datasetId: synID of the dataset for the manifest
3006        existingTableId: synId of the table currently exising on synapse (if there is one)
3007        restrict: bool, whether or not the manifest contains sensitive data that will need additional access restrictions
3008        synapse_entity_tracker: Tracker for a pull-through cache of Synapse entities
3009
3010        """
3011        self.synStore = synStore
3012        self.tableToLoad = tableToLoad
3013        self.tableName = tableName
3014        self.datasetId = datasetId
3015        self.existingTableId = existingTableId
3016        self.restrict = restrict
3017        self.synapse_entity_tracker = synapse_entity_tracker or SynapseEntityTracker()
3018
3019    @tracer.start_as_current_span("TableOperations::createTable")
3020    def createTable(
3021        self,
3022        columnTypeDict: dict = None,
3023        specifySchema: bool = True,
3024    ):
3025        """
3026        Method to create a table from a metadata manifest and upload it to synapse
3027
3028        Args:
3029            columnTypeDict: dictionary schema for table columns: type, size, etc
3030            specifySchema: to specify a specific schema for the table format
3031
3032        Returns:
3033            table.schema.id: synID of the newly created table
3034        """
3035        datasetEntity = self.synapse_entity_tracker.get(
3036            synapse_id=self.datasetId, syn=self.synStore.syn, download_file=False
3037        )
3038        datasetName = datasetEntity.name
3039        table_schema_by_cname = self.synStore._get_table_schema_by_cname(columnTypeDict)
3040
3041        if not self.tableName:
3042            self.tableName = datasetName + "table"
3043        datasetParentProject = self.synStore.getDatasetProject(self.datasetId)
3044        if specifySchema:
3045            if columnTypeDict == {}:
3046                logger.error("Did not provide a columnTypeDict.")
3047            # create list of columns:
3048            cols = []
3049            for col in self.tableToLoad.columns:
3050                if col in table_schema_by_cname:
3051                    col_type = table_schema_by_cname[col]["columnType"]
3052                    max_size = (
3053                        table_schema_by_cname[col]["maximumSize"]
3054                        if "maximumSize" in table_schema_by_cname[col].keys()
3055                        else 100
3056                    )
3057                    max_list_len = 250
3058                    if max_size and max_list_len:
3059                        cols.append(
3060                            Column(
3061                                name=col,
3062                                columnType=col_type,
3063                                maximumSize=max_size,
3064                                maximumListLength=max_list_len,
3065                            )
3066                        )
3067                    elif max_size:
3068                        cols.append(
3069                            Column(name=col, columnType=col_type, maximumSize=max_size)
3070                        )
3071                    else:
3072                        cols.append(Column(name=col, columnType=col_type))
3073                else:
3074                    # TODO add warning that the given col was not found and it's max size is set to 100
3075                    cols.append(Column(name=col, columnType="STRING", maximumSize=100))
3076            schema = Schema(
3077                name=self.tableName, columns=cols, parent=datasetParentProject
3078            )
3079            table = Table(schema, self.tableToLoad)
3080            table = self.synStore.syn.store(table, isRestricted=self.restrict)
3081            # Commented out until https://sagebionetworks.jira.com/browse/PLFM-8605 is resolved
3082            # self.synapse_entity_tracker.add(synapse_id=table.schema.id, entity=table.schema)
3083            self.synapse_entity_tracker.remove(synapse_id=table.schema.id)
3084            return table.schema.id
3085        else:
3086            # For just uploading the tables to synapse using default
3087            # column types.
3088            table = build_table(self.tableName, datasetParentProject, self.tableToLoad)
3089            table = self.synStore.syn.store(table, isRestricted=self.restrict)
3090            # Commented out until https://sagebionetworks.jira.com/browse/PLFM-8605 is resolved
3091            # self.synapse_entity_tracker.add(synapse_id=table.schema.id, entity=table.schema)
3092            self.synapse_entity_tracker.remove(synapse_id=table.schema.id)
3093            return table.schema.id
3094
3095    @tracer.start_as_current_span("TableOperations::replaceTable")
3096    def replaceTable(
3097        self,
3098        specifySchema: bool = True,
3099        columnTypeDict: dict = None,
3100    ):
3101        """
3102        Method to replace an existing table on synapse with metadata from a new manifest
3103
3104        Args:
3105            specifySchema: to infer a schema for the table format
3106            columnTypeDict: dictionary schema for table columns: type, size, etc
3107
3108        Returns:
3109           existingTableId: synID of the already existing table that had its metadata replaced
3110        """
3111        datasetEntity = self.synapse_entity_tracker.get(
3112            synapse_id=self.datasetId, syn=self.synStore.syn, download_file=False
3113        )
3114
3115        datasetName = datasetEntity.name
3116        table_schema_by_cname = self.synStore._get_table_schema_by_cname(columnTypeDict)
3117        existing_table, existing_results = self.synStore.get_synapse_table(
3118            self.existingTableId
3119        )
3120        # remove rows
3121        self.synStore.syn.delete(existing_results)
3122        # Data changes such as removing all rows causes the eTag to change.
3123        self.synapse_entity_tracker.remove(synapse_id=self.existingTableId)
3124        # wait for row deletion to finish on synapse before getting empty table
3125        sleep(10)
3126
3127        # removes all current columns
3128        current_table = self.synapse_entity_tracker.get(
3129            synapse_id=self.existingTableId, syn=self.synStore.syn, download_file=False
3130        )
3131
3132        current_columns = self.synStore.syn.getTableColumns(current_table)
3133        for col in current_columns:
3134            current_table.removeColumn(col)
3135
3136        if not self.tableName:
3137            self.tableName = datasetName + "table"
3138
3139        # Process columns according to manifest entries
3140        table_schema_by_cname = self.synStore._get_table_schema_by_cname(columnTypeDict)
3141        datasetParentProject = self.synStore.getDatasetProject(self.datasetId)
3142        if specifySchema:
3143            if columnTypeDict == {}:
3144                logger.error("Did not provide a columnTypeDict.")
3145            # create list of columns:
3146            cols = []
3147
3148            for col in self.tableToLoad.columns:
3149                if col in table_schema_by_cname:
3150                    col_type = table_schema_by_cname[col]["columnType"]
3151                    max_size = (
3152                        table_schema_by_cname[col]["maximumSize"]
3153                        if "maximumSize" in table_schema_by_cname[col].keys()
3154                        else 100
3155                    )
3156                    max_list_len = 250
3157                    if max_size and max_list_len:
3158                        cols.append(
3159                            Column(
3160                                name=col,
3161                                columnType=col_type,
3162                                maximumSize=max_size,
3163                                maximumListLength=max_list_len,
3164                            )
3165                        )
3166                    elif max_size:
3167                        cols.append(
3168                            Column(name=col, columnType=col_type, maximumSize=max_size)
3169                        )
3170                    else:
3171                        cols.append(Column(name=col, columnType=col_type))
3172                else:
3173                    # TODO add warning that the given col was not found and it's max size is set to 100
3174                    cols.append(Column(name=col, columnType="STRING", maximumSize=100))
3175
3176            # adds new columns to schema
3177            for col in cols:
3178                current_table.addColumn(col)
3179            table_result = self.synStore.syn.store(
3180                current_table, isRestricted=self.restrict
3181            )
3182            # Commented out until https://sagebionetworks.jira.com/browse/PLFM-8605 is resolved
3183            # self.synapse_entity_tracker.add(synapse_id=table_result.schema.id, entity=table_result.schema)
3184            self.synapse_entity_tracker.remove(synapse_id=table_result.id)
3185
3186            # wait for synapse store to finish
3187            sleep(1)
3188
3189            # build schema and table from columns and store with necessary restrictions
3190            schema = Schema(
3191                name=self.tableName, columns=cols, parent=datasetParentProject
3192            )
3193            schema.id = self.existingTableId
3194            table = Table(schema, self.tableToLoad, etag=existing_results.etag)
3195            table = self.synStore.syn.store(table, isRestricted=self.restrict)
3196            # Commented out until https://sagebionetworks.jira.com/browse/PLFM-8605 is resolved
3197            # self.synapse_entity_tracker.add(synapse_id=table.schema.id, entity=table.schema)
3198            self.synapse_entity_tracker.remove(synapse_id=table.schema.id)
3199        else:
3200            logging.error("Must specify a schema for table replacements")
3201
3202        # remove system metadata from manifest
3203        existing_table.drop(columns=["ROW_ID", "ROW_VERSION"], inplace=True)
3204        return self.existingTableId
3205
3206    @tracer.start_as_current_span("TableOperations::_get_auth_token")
3207    def _get_auth_token(
3208        self,
3209    ):
3210        authtoken = None
3211
3212        # Get access token from environment variable if available
3213        # Primarily useful for testing environments, with other possible usefulness for containers
3214        env_access_token = os.getenv("SYNAPSE_ACCESS_TOKEN")
3215        if env_access_token:
3216            authtoken = env_access_token
3217            return authtoken
3218
3219        # Get token from authorization header
3220        # Primarily useful for API endpoint functionality
3221        if "Authorization" in self.synStore.syn.default_headers:
3222            authtoken = self.synStore.syn.default_headers["Authorization"].split(
3223                "Bearer "
3224            )[-1]
3225            return authtoken
3226
3227        # retrive credentials from synapse object
3228        # Primarily useful for local users, could only be stored here when a .synapseConfig file is used, but including to be safe
3229        synapse_object_creds = self.synStore.syn.credentials
3230        if hasattr(synapse_object_creds, "_token"):
3231            authtoken = synapse_object_creds.secret
3232
3233        # Try getting creds from .synapseConfig file if it exists
3234        # Primarily useful for local users. Seems to correlate with credentials stored in synaspe object when logged in
3235        if os.path.exists(CONFIG.synapse_configuration_path):
3236            config = get_config_file(CONFIG.synapse_configuration_path)
3237
3238            # check which credentials are provided in file
3239            if config.has_option("authentication", "authtoken"):
3240                authtoken = config.get("authentication", "authtoken")
3241
3242        # raise error if required credentials are not found
3243        if not authtoken:
3244            raise NameError(
3245                "authtoken credentials could not be found in the environment, synapse object, or the .synapseConfig file"
3246            )
3247
3248        return authtoken
3249
3250    @tracer.start_as_current_span("TableOperations::upsertTable")
3251    def upsertTable(self, dmge: DataModelGraphExplorer):
3252        """
3253        Method to upsert rows from a new manifest into an existing table on synapse
3254        For upsert functionality to work, primary keys must follow the naming convention of <componenet>_id
3255        `-tm upsert` should be used for initial table uploads if users intend to upsert into them at a later time; using 'upsert' at creation will generate the metadata necessary for upsert functionality.
3256        Currently it is required to use -dl/--use_display_label with table upserts.
3257
3258
3259        Args:
3260            dmge: DataModelGraphExplorer instance
3261
3262        Returns:
3263           existingTableId: synID of the already existing table that had its metadata replaced
3264        """
3265
3266        authtoken = self._get_auth_token()
3267
3268        synapseDB = SynapseDatabase(
3269            auth_token=authtoken,
3270            project_id=self.synStore.getDatasetProject(self.datasetId),
3271            syn=self.synStore.syn,
3272            synapse_entity_tracker=self.synapse_entity_tracker,
3273        )
3274
3275        try:
3276            # Try performing upsert
3277            synapseDB.upsert_table_rows(
3278                table_name=self.tableName, data=self.tableToLoad
3279            )
3280        except SynapseHTTPError as ex:
3281            # If error is raised because Table has old `Uuid` column and not new `Id` column, then handle and re-attempt upload
3282            if "Id is not a valid column name or id" in str(ex):
3283                self._update_table_uuid_column(dmge)
3284                synapseDB.upsert_table_rows(
3285                    table_name=self.tableName, data=self.tableToLoad
3286                )
3287            # Raise if other error
3288            else:
3289                raise ex
3290
3291        return self.existingTableId
3292
3293    @tracer.start_as_current_span("TableOperations::_update_table_uuid_column")
3294    def _update_table_uuid_column(
3295        self,
3296        dmge: DataModelGraphExplorer,
3297    ) -> None:
3298        """Removes the `Uuid` column when present, and relpaces with an `Id` column
3299        Used to enable backwards compatability for manifests using the old `Uuid` convention
3300
3301        Args:
3302            dmge: DataModelGraphExplorer instance
3303
3304        Returns:
3305            None
3306        """
3307
3308        # Get the columns of the schema
3309        schema = self.synapse_entity_tracker.get(
3310            synapse_id=self.existingTableId, syn=self.synStore.syn, download_file=False
3311        )
3312
3313        cols = self.synStore.syn.getTableColumns(schema)
3314
3315        # Iterate through columns until `Uuid` column is found
3316        for col in cols:
3317            if col.name.lower() == "uuid":
3318                # See if schema has `Uuid` column specified
3319                try:
3320                    uuid_col_in_schema = dmge.is_class_in_schema(col.name)
3321                except KeyError:
3322                    uuid_col_in_schema = False
3323
3324                # If there is, then create a new `Id` column from scratch
3325                if uuid_col_in_schema:
3326                    new_col = Column(columnType="STRING", maximumSize=64, name="Id")
3327                    schema.addColumn(new_col)
3328                    schema = self.synStore.syn.store(schema)
3329                    # self.synapse_entity_tracker.add(synapse_id=schema.id, entity=schema)
3330                    # Commented out until https://sagebionetworks.jira.com/browse/PLFM-8605 is resolved
3331                    self.synapse_entity_tracker.remove(synapse_id=schema.id)
3332                # If there is not, then use the old `Uuid` column as a basis for the new `Id` column
3333                else:
3334                    # Build ColumnModel that will be used for new column
3335                    id_column = Column(
3336                        name="Id",
3337                        columnType="STRING",
3338                        maximumSize=64,
3339                        defaultValue=None,
3340                        maximumListLength=1,
3341                    )
3342                    new_col_response = self.synStore.syn.store(id_column)
3343
3344                    # Define columnChange body
3345                    columnChangeDict = {
3346                        "concreteType": "org.sagebionetworks.repo.model.table.TableSchemaChangeRequest",
3347                        "entityId": self.existingTableId,
3348                        "changes": [
3349                            {
3350                                "oldColumnId": col["id"],
3351                                "newColumnId": new_col_response["id"],
3352                            }
3353                        ],
3354                    }
3355
3356                    self.synStore.syn._async_table_update(
3357                        table=self.existingTableId,
3358                        changes=[columnChangeDict],
3359                        wait=False,
3360                    )
3361                break
3362
3363        return
3364
3365    @tracer.start_as_current_span("TableOperations::updateTable")
3366    def updateTable(
3367        self,
3368        update_col: str = "Id",
3369    ):
3370        """
3371        Method to update an existing table with a new column
3372
3373        Args:
3374            updateCol: column to index the old and new tables on
3375
3376        Returns:
3377           existingTableId: synID of the already existing table that had its metadata replaced
3378        """
3379        existing_table, existing_results = self.synStore.get_synapse_table(
3380            self.existingTableId
3381        )
3382
3383        self.tableToLoad = update_df(existing_table, self.tableToLoad, update_col)
3384        # store table with existing etag data and impose restrictions as appropriate
3385        table_result = self.synStore.syn.store(
3386            Table(self.existingTableId, self.tableToLoad, etag=existing_results.etag),
3387            isRestricted=self.restrict,
3388        )
3389        # We cannot store the Table to the `synapse_entity_tracker` because there is
3390        # not `Schema` on the table object. The above `.store()` function call would
3391        # also update the ETag of the entity within Synapse. Remove it from the tracker
3392        # and re-retrieve it later on if needed again.
3393        self.synapse_entity_tracker.remove(synapse_id=table_result.tableId)
3394
3395        return self.existingTableId
3396
3397
3398class DatasetFileView:
3399    """Helper class to create temporary dataset file views.
3400    This class can be used in conjunction with a 'with' statement.
3401    This will ensure that the file view is deleted automatically.
3402    See SynapseStorage.getDatasetAnnotationsBatch for example usage.
3403    """
3404
3405    def __init__(
3406        self,
3407        datasetId: str,
3408        synapse: Synapse,
3409        name: str = None,
3410        temporary: bool = True,
3411        parentId: str = None,
3412    ) -> None:
3413        """Create a file view scoped to a dataset folder.
3414
3415        Args:
3416            datasetId (str): Synapse ID for a dataset folder/project.
3417            synapse (Synapse): Used for Synapse requests.
3418            name (str): Name of the file view (temporary or not).
3419            temporary (bool): Whether to delete the file view on exit
3420                of either a 'with' statement or Python entirely.
3421            parentId (str, optional): Synapse ID specifying where to
3422                store the file view. Defaults to datasetId.
3423        """
3424
3425        self.datasetId = datasetId
3426        self.synapse = synapse
3427        self.is_temporary = temporary
3428
3429        if name is None:
3430            self.name = f"schematic annotation file view for {self.datasetId}"
3431
3432        if self.is_temporary:
3433            uid = secrets.token_urlsafe(5)
3434            self.name = f"{self.name} - UID {uid}"
3435
3436        # TODO: Allow a DCC admin to configure a "universal parent"
3437        #       Such as a Synapse project writeable by everyone.
3438        self.parentId = datasetId if parentId is None else parentId
3439
3440        # TODO: Create local sharing setting to hide from everyone else
3441        view_schema = EntityViewSchema(
3442            name=self.name,
3443            parent=self.parentId,
3444            scopes=self.datasetId,
3445            includeEntityTypes=[EntityViewType.FILE, EntityViewType.FOLDER],
3446            addDefaultViewColumns=False,
3447            addAnnotationColumns=True,
3448        )
3449
3450        # TODO: Handle failure due to insufficient permissions by
3451        #       creating a temporary new project to store view
3452        self.view_schema = self.synapse.store(view_schema)
3453
3454        # These are filled in after calling `self.query()`
3455        self.results = None
3456        self.table = None
3457
3458        # Ensure deletion of the file view (last resort)
3459        if self.is_temporary:
3460            atexit.register(self.delete)
3461
3462    def __enter__(self):
3463        """Return file view when entering 'with' statement."""
3464        return self
3465
3466    def __exit__(self, exc_type, exc_value, traceback):
3467        """Delete file view when exiting 'with' statement."""
3468        if self.is_temporary:
3469            self.delete()
3470
3471    def delete(self):
3472        """Delete the file view on Synapse without deleting local table."""
3473        if self.view_schema is not None:
3474            self.synapse.delete(self.view_schema)
3475            self.view_schema = None
3476
3477    def query(self, tidy=True, force=False):
3478        """Retrieve file view as a data frame (raw format sans index)."""
3479        if self.table is None or force:
3480            fileview_id = self.view_schema["id"]
3481            self.results = self.synapse.tableQuery(f"select * from {fileview_id}")
3482            self.table = self.results.asDataFrame(
3483                rowIdAndVersionInIndex=False,
3484                na_values=STR_NA_VALUES_FILTERED,
3485                keep_default_na=False,
3486            )
3487        if tidy:
3488            self.tidy_table()
3489        return self.table
3490
3491    def tidy_table(self):
3492        """Convert raw file view data frame into more usable format."""
3493        assert self.table is not None, "Must call `self.query()` first."
3494        self._fix_default_columns()
3495        self._fix_list_columns()
3496        self._fix_int_columns()
3497        return self.table
3498
3499    def _fix_default_columns(self):
3500        """Rename default columns to match schematic expectations."""
3501
3502        # Drop ROW_VERSION column if present
3503        if "ROW_VERSION" in self.table:
3504            del self.table["ROW_VERSION"]
3505
3506        # Rename id column to entityId and set as data frame index
3507        if "ROW_ID" in self.table:
3508            self.table["entityId"] = "syn" + self.table["ROW_ID"].astype(str)
3509            self.table = self.table.set_index("entityId", drop=False)
3510            del self.table["ROW_ID"]
3511
3512        # Rename ROW_ETAG column to eTag and place at end of data frame
3513        if "ROW_ETAG" in self.table:
3514            row_etags = self.table.pop("ROW_ETAG")
3515
3516            # eTag column may already present if users annotated data without submitting manifest
3517            # we're only concerned with the new values and not the existing ones
3518            if "eTag" in self.table:
3519                del self.table["eTag"]
3520
3521            self.table.insert(len(self.table.columns), "eTag", row_etags)
3522
3523        return self.table
3524
3525    def _get_columns_of_type(self, types):
3526        """Helper function to get list of columns of a given type(s)."""
3527        matching_columns = []
3528        for header in self.results.headers:
3529            if header.columnType in types:
3530                matching_columns.append(header.name)
3531        return matching_columns
3532
3533    def _fix_list_columns(self):
3534        """Fix formatting of list-columns."""
3535        list_types = {"STRING_LIST", "INTEGER_LIST", "BOOLEAN_LIST"}
3536        list_columns = self._get_columns_of_type(list_types)
3537        for col in list_columns:
3538            self.table[col] = self.table[col].apply(lambda x: ", ".join(x))
3539        return self.table
3540
3541    def _fix_int_columns(self):
3542        """Ensure that integer-columns are actually integers."""
3543        int_columns = self._get_columns_of_type({"INTEGER"})
3544        for col in int_columns:
3545            # Coercing to string because NaN is a floating point value
3546            # and cannot exist alongside integers in a column
3547            def to_int_fn(x):
3548                return "" if np.isnan(x) else str(int(x))
3549
3550            self.table[col] = self.table[col].apply(to_int_fn)
3551        return self.table