cirro.sdk.portal

  1from cirro_api_client.v1.models import Executor
  2
  3from cirro.cirro_client import CirroApi
  4from cirro.sdk.dataset import DataPortalDataset
  5from cirro.sdk.exceptions import DataPortalAssetNotFound
  6from cirro.sdk.process import DataPortalProcess, DataPortalProcesses
  7from cirro.sdk.project import DataPortalProject, DataPortalProjects
  8from cirro.sdk.reference_type import DataPortalReferenceType, DataPortalReferenceTypes
  9
 10
 11class DataPortal:
 12    """
 13    Helper functions for exploring the Projects, Datasets, Samples, and Files
 14    available in the Data Portal.
 15    """
 16
 17    def __init__(self, base_url: str = None, client: CirroApi = None):
 18        """
 19        Set up the DataPortal object, establishing an authenticated connection.
 20
 21        Args:
 22            base_url (str): Optional base URL of the Cirro instance
 23             (if not provided, it uses the `CIRRO_BASE_URL` environment variable, or the config file)
 24            client (`cirro.cirro_client.CirroApi`): Optional pre-configured client
 25
 26        Example:
 27        ```python
 28        from cirro import DataPortal
 29
 30        Portal = DataPortal(base_url="app.cirro.bio")
 31        portal.list_projects()
 32        ```
 33        """
 34
 35        if client is not None:
 36            self._client = client
 37
 38        # Set up default client if not provided
 39        else:
 40            self._client = CirroApi(base_url=base_url)
 41
 42    def list_projects(self) -> DataPortalProjects:
 43        """List all the projects available in the Data Portal."""
 44
 45        return DataPortalProjects(
 46            [
 47                DataPortalProject(proj, self._client)
 48                for proj in self._client.projects.list()
 49            ]
 50        )
 51
 52    def get_project_by_name(self, name: str = None) -> DataPortalProject:
 53        """Return the project with the specified name."""
 54
 55        return self.list_projects().get_by_name(name)
 56
 57    def get_project_by_id(self, _id: str = None) -> DataPortalProject:
 58        """Return the project with the specified id."""
 59
 60        return self.list_projects().get_by_id(_id)
 61
 62    def get_project(self, project: str = None) -> DataPortalProject:
 63        """
 64        Return a project identified by ID or name.
 65
 66        Args:
 67            project (str): ID or name of project
 68
 69        Returns:
 70            `from cirro.sdk.project import DataPortalProject`
 71        """
 72        try:
 73            return self.get_project_by_id(project)
 74        except DataPortalAssetNotFound:
 75            return self.get_project_by_name(project)
 76
 77    def get_dataset(self, project: str = None, dataset: str = None) -> DataPortalDataset:
 78        """
 79        Return a dataset identified by ID or name.
 80
 81        Args:
 82            project (str): ID or name of project
 83            dataset (str): ID or name of dataset
 84
 85        Returns:
 86            `cirro.sdk.dataset.DataPortalDataset`
 87
 88            ```python
 89            from cirro import DataPortal()
 90            portal = DataPortal()
 91            dataset = portal.get_dataset(
 92                project="id-or-name-of-project",
 93                dataset="id-or-name-of-dataset"
 94            )
 95            ```
 96        """
 97        try:
 98            project: DataPortalProject = self.get_project_by_id(project)
 99        except DataPortalAssetNotFound:
100            project: DataPortalProject = self.get_project_by_name(project)
101
102        try:
103            return project.get_dataset_by_id(dataset)
104        except DataPortalAssetNotFound:
105            return project.get_dataset_by_name(dataset)
106
107    def list_processes(self, ingest=False) -> DataPortalProcesses:
108        """
109        List all the processes available in the Data Portal.
110        By default, only list non-ingest processes (those which can be run on existing datasets).
111        To list the processes which can be used to upload datasets, use `ingest = True`.
112
113        Args:
114            ingest (bool): If True, only list those processes which can be used to ingest datasets directly
115        """
116
117        return DataPortalProcesses(
118            [
119                DataPortalProcess(p, self._client)
120                for p in self._client.processes.list()
121                if not ingest or p.executor == Executor.INGEST
122            ]
123        )
124
125    def get_process_by_name(self, name: str, ingest=False) -> DataPortalProcess:
126        """
127        Return the process with the specified name.
128
129        Args:
130            name (str): Name of process
131        """
132
133        return self.list_processes(ingest=ingest).get_by_name(name)
134
135    def get_process_by_id(self, id: str, ingest=False) -> DataPortalProcess:
136        """
137        Return the process with the specified id
138
139        Args:
140            id (str): ID of process
141        """
142
143        return self.list_processes(ingest=ingest).get_by_id(id)
144
145    def list_reference_types(self) -> DataPortalReferenceTypes:
146        """
147        Return the list of all available reference types
148        """
149
150        return DataPortalReferenceTypes(
151            [
152                DataPortalReferenceType(ref)
153                for ref in self._client.references.get_types()
154            ]
155        )