1  import warnings
  2  from typing import List
  3  from typing import Optional
  4  from typing import Union
  5  
  6  import pandas as pd
  7  import uuid6
  8  
  9  from evidently.legacy.pipeline.column_mapping import ColumnMapping
 10  from evidently.legacy.suite.base_suite import Snapshot
 11  from evidently.legacy.ui.base import Project
 12  from evidently.legacy.ui.datasets import DatasetSourceType
 13  from evidently.legacy.ui.managers.projects import ProjectManager
 14  from evidently.legacy.ui.type_aliases import STR_UUID
 15  from evidently.legacy.ui.type_aliases import ZERO_UUID
 16  from evidently.legacy.ui.type_aliases import DatasetID
 17  from evidently.legacy.ui.type_aliases import OrgID
 18  from evidently.legacy.ui.type_aliases import TeamID
 19  from evidently.legacy.ui.type_aliases import UserID
 20  from evidently.legacy.ui.workspace.base import WorkspaceBase
 21  from evidently.legacy.utils.sync import async_to_sync
 22  
 23  
 24  def _team_id_deprecation_message(team_id: Optional[TeamID]):
 25      if team_id is not None:
 26          warnings.warn(
 27              "The 'team_id' parameter is deprecated and will be removed in a future version. "
 28              "Please use 'org_id' instead.",
 29              DeprecationWarning,
 30              stacklevel=2,
 31          )
 32  
 33  
 34  class WorkspaceView(WorkspaceBase):
 35      def __init__(
 36          self,
 37          user_id: Optional[UserID],
 38          project_manager: ProjectManager,
 39      ):
 40          self.project_manager = project_manager
 41          self.user_id = user_id or ZERO_UUID
 42  
 43      def create_project(
 44          self,
 45          name: str,
 46          description: Optional[str] = None,
 47          team_id: Optional[TeamID] = None,
 48          org_id: Optional[OrgID] = None,
 49      ) -> Project:
 50          _team_id_deprecation_message(team_id)
 51          return async_to_sync(
 52              self.project_manager.create_project(
 53                  name, user_id=self.user_id, team_id=team_id, description=description, org_id=org_id
 54              )
 55          )
 56  
 57      def add_project(
 58          self, project: Project, team_id: Optional[TeamID] = None, org_id: Optional[OrgID] = None
 59      ) -> Project:
 60          _team_id_deprecation_message(team_id)
 61          project = async_to_sync(
 62              self.project_manager.add_project(project, user_id=self.user_id, team_id=team_id, org_id=org_id)
 63          )
 64          return project
 65  
 66      def get_project(self, project_id: STR_UUID) -> Optional[Project]:
 67          if isinstance(project_id, str):
 68              project_id = uuid6.UUID(project_id)
 69          return async_to_sync(self.project_manager.get_project(self.user_id, project_id))
 70  
 71      def delete_project(self, project_id: STR_UUID):
 72          if isinstance(project_id, str):
 73              project_id = uuid6.UUID(project_id)
 74          async_to_sync(self.project_manager.delete_project(self.user_id, project_id))
 75  
 76      def list_projects(self, team_id: Optional[TeamID] = None, org_id: Optional[OrgID] = None) -> List[Project]:
 77          _team_id_deprecation_message(team_id)
 78          return async_to_sync(
 79              self.project_manager.list_projects(self.user_id, team_id or ZERO_UUID, org_id or ZERO_UUID)
 80          )
 81  
 82      def add_snapshot(self, project_id: STR_UUID, snapshot: Snapshot):
 83          if isinstance(project_id, str):
 84              project_id = uuid6.UUID(project_id)
 85          async_to_sync(self.project_manager.add_snapshot(self.user_id, project_id, snapshot))
 86  
 87      def delete_snapshot(self, project_id: STR_UUID, snapshot_id: STR_UUID):
 88          if isinstance(project_id, str):
 89              project_id = uuid6.UUID(project_id)
 90          if isinstance(snapshot_id, str):
 91              snapshot_id = uuid6.UUID(snapshot_id)
 92          async_to_sync(self.project_manager.delete_snapshot(self.user_id, project_id, snapshot_id))
 93  
 94      def search_project(
 95          self, project_name: str, team_id: Optional[TeamID] = None, org_id: Optional[OrgID] = None
 96      ) -> List[Project]:
 97          _team_id_deprecation_message(team_id)
 98          return async_to_sync(
 99              self.project_manager.search_project(self.user_id, project_name, team_id or ZERO_UUID, org_id or ZERO_UUID)
100          )
101  
102      def add_dataset(
103          self,
104          data_or_path: Union[str, pd.DataFrame],
105          name: str,
106          project_id: STR_UUID,
107          description: Optional[str] = None,
108          column_mapping: Optional[ColumnMapping] = None,
109          dataset_source: DatasetSourceType = DatasetSourceType.file,
110      ) -> DatasetID:
111          raise NotImplementedError("Adding datasets is not supported yet")
112  
113  
114  class LocalWorkspaceView(WorkspaceView):
115      def __init__(self, path: str):
116          from evidently.legacy.ui.storage.local import create_local_project_manager
117  
118          self.path = path
119          super().__init__(None, create_local_project_manager(path=path, autorefresh=False))
120  
121      @classmethod
122      def create(cls, path: str):
123          return LocalWorkspaceView(path)
124  
125      def refresh(self):
126          from evidently.legacy.ui.storage.local import create_local_project_manager
127  
128          self.project_manager = create_local_project_manager(path=self.path, autorefresh=False)
129  
130  
131  Workspace = LocalWorkspaceView