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