projects.py
1 import asyncio 2 import datetime 3 import json 4 from typing import Callable 5 from typing import Dict 6 from typing import List 7 from typing import Optional 8 from typing import Sequence 9 from typing import Union 10 11 from litestar import Response 12 from litestar import Router 13 from litestar import delete 14 from litestar import get 15 from litestar import post 16 from litestar.di import Provide 17 from litestar.exceptions import HTTPException 18 from litestar.params import Dependency 19 from litestar.params import Parameter 20 from typing_extensions import Annotated 21 22 from evidently.legacy.report.report import METRIC_GENERATORS 23 from evidently.legacy.report.report import METRIC_PRESETS 24 from evidently.legacy.suite.base_suite import Snapshot 25 from evidently.legacy.test_suite.test_suite import TEST_GENERATORS 26 from evidently.legacy.test_suite.test_suite import TEST_PRESETS 27 from evidently.legacy.ui.api.models import DashboardInfoModel 28 from evidently.legacy.ui.api.models import ReportModel 29 from evidently.legacy.ui.api.models import TestSuiteModel 30 from evidently.legacy.ui.base import Project 31 from evidently.legacy.ui.base import SnapshotMetadata 32 from evidently.legacy.ui.dashboards.base import DashboardPanel 33 from evidently.legacy.ui.dashboards.reports import DashboardPanelCounter 34 from evidently.legacy.ui.dashboards.reports import DashboardPanelDistribution 35 from evidently.legacy.ui.dashboards.reports import DashboardPanelHistogram 36 from evidently.legacy.ui.dashboards.reports import DashboardPanelPlot 37 from evidently.legacy.ui.dashboards.test_suites import DashboardPanelTestSuite 38 from evidently.legacy.ui.dashboards.test_suites import DashboardPanelTestSuiteCounter 39 from evidently.legacy.ui.managers.projects import ProjectManager 40 from evidently.legacy.ui.type_aliases import OrgID 41 from evidently.legacy.ui.type_aliases import ProjectID 42 from evidently.legacy.ui.type_aliases import SnapshotID 43 from evidently.legacy.ui.type_aliases import TeamID 44 from evidently.legacy.ui.type_aliases import UserID 45 from evidently.legacy.utils import NumpyEncoder 46 47 48 async def path_project_dependency( 49 project_id: ProjectID, 50 project_manager: Annotated[ProjectManager, Dependency(skip_validation=True)], 51 user_id: UserID, 52 ): 53 project = await project_manager.get_project(user_id, project_id) 54 if project is None: 55 raise HTTPException(status_code=404, detail="project not found") 56 return project 57 58 59 @get("/{project_id:uuid}/reports") 60 async def list_reports( 61 project: Annotated[Project, Dependency()], 62 log_event: Callable, 63 ) -> List[ReportModel]: 64 reports = [ 65 ReportModel.from_snapshot(s) 66 for s in await project.list_snapshots_async(include_test_suites=False) 67 if s.is_report 68 ] 69 log_event("list_reports", reports_count=len(reports)) 70 return reports 71 72 73 @get("/{project_id:uuid}/test_suites") 74 async def list_test_suites( 75 project: Annotated[Project, Dependency()], 76 log_event: Callable, 77 ) -> List[TestSuiteModel]: 78 log_event("list_test_suites") 79 return [ 80 TestSuiteModel.from_snapshot(s) 81 for s in await project.list_snapshots_async(include_reports=False) 82 if not s.is_report 83 ] 84 85 86 @get("/{project_id:uuid}/snapshots") 87 async def list_snapshots( 88 project: Annotated[Project, Dependency()], 89 project_manager: Annotated[ProjectManager, Dependency(skip_validation=True)], 90 log_event: Callable, 91 user_id: UserID, 92 ) -> List[SnapshotMetadata]: 93 snapshots = await project_manager.list_snapshots(user_id, project.id) 94 log_event("list_snapshots", reports_count=len(snapshots)) 95 return snapshots 96 97 98 @get("/") 99 async def list_projects( 100 project_manager: Annotated[ProjectManager, Dependency(skip_validation=True)], 101 log_event: Callable, 102 user_id: UserID, 103 team_id: Annotated[Optional[TeamID], Parameter(title="filter by team")] = None, 104 org_id: Annotated[Optional[OrgID], Parameter(title="filter by org")] = None, 105 ) -> Sequence[Project]: 106 projects = await project_manager.list_projects(user_id, team_id, org_id) 107 log_event("list_projects", project_count=len(projects)) 108 return projects 109 110 111 @get("/{project_id:uuid}/info") 112 async def get_project_info( 113 project: Annotated[Project, Dependency()], 114 log_event: Callable, 115 ) -> Project: 116 log_event("get_project_info") 117 return project 118 119 120 @get("/search/{project_name:str}") 121 async def search_projects( 122 project_name: Annotated[str, Parameter(title="Name of the project to search")], 123 project_manager: Annotated[ProjectManager, Dependency(skip_validation=True)], 124 log_event: Callable, 125 user_id: UserID, 126 team_id: Annotated[Optional[TeamID], Parameter(title="filter by team")] = None, 127 org_id: Annotated[Optional[OrgID], Parameter(title="filter by org")] = None, 128 ) -> List[Project]: 129 log_event("search_projects") 130 return await project_manager.search_project(user_id, team_id=team_id, org_id=org_id, project_name=project_name) 131 132 133 @post("/{project_id:uuid}/info") 134 async def update_project_info( 135 project: Annotated[Project, Dependency()], 136 data: Project, 137 project_manager: Annotated[ProjectManager, Dependency(skip_validation=True)], 138 log_event: Callable, 139 user_id: UserID, 140 ) -> Project: 141 data.id = project.id 142 await project_manager.update_project(user_id, data) 143 log_event("update_project_info") 144 return project 145 146 147 @get("/{project_id:uuid}/reload") 148 async def reload_project_snapshots( 149 project: Annotated[Project, Dependency()], 150 log_event: Callable, 151 ) -> None: 152 await project.reload_async(reload_snapshots=True) 153 log_event("reload_project_snapshots") 154 155 156 async def path_snapshot_metadata_dependency( 157 project: Annotated[Project, Dependency()], 158 snapshot_id: SnapshotID, 159 ): 160 snapshot = await project.get_snapshot_metadata_async(snapshot_id) 161 if snapshot is None: 162 raise HTTPException(status_code=404, detail="Snapshot not found") 163 return snapshot 164 165 166 @get( 167 "/{project_id:uuid}/{snapshot_id:uuid}/graphs_data/{graph_id:str}", 168 ) 169 async def get_snapshot_graph_data( 170 snapshot_metadata: Annotated[SnapshotMetadata, Dependency()], 171 graph_id: Annotated[str, Parameter(title="id of graph in snapshot")], 172 log_event: Callable, 173 ) -> str: 174 graph = (await snapshot_metadata.get_additional_graphs()).get(graph_id) 175 if graph is None: 176 raise HTTPException(status_code=404, detail="Graph not found") 177 log_event("get_snapshot_graph_data") 178 return json.dumps(graph.dict() if not isinstance(graph, dict) else graph, cls=NumpyEncoder) 179 180 181 @get("/{project_id:uuid}/{snapshot_id:uuid}/download") 182 async def get_snapshot_download( 183 snapshot_metadata: Annotated[SnapshotMetadata, Dependency()], 184 log_event: Callable, 185 report_format: str = "html", 186 ) -> Response: 187 report = await snapshot_metadata.as_report_base() 188 if report_format == "html": 189 return Response( 190 report.get_html(), 191 headers={"content-disposition": f"attachment;filename={snapshot_metadata.id}.html"}, 192 ) 193 if report_format == "json": 194 return Response( 195 report.json(), 196 headers={"content-disposition": f"attachment;filename={snapshot_metadata.id}.json"}, 197 ) 198 log_event("get_snapshot_download") 199 raise HTTPException(status_code=400, detail=f"Unknown format {report_format}") 200 201 202 @get("/{project_id:uuid}/{snapshot_id:uuid}/data") 203 async def get_snapshot_data( 204 snapshot_metadata: Annotated[SnapshotMetadata, Dependency()], 205 log_event: Callable, 206 ) -> str: 207 dashboard_info, snapshot = await asyncio.gather(snapshot_metadata.get_dashboard_info(), snapshot_metadata.load()) 208 info = DashboardInfoModel.from_dashboard_info(dashboard_info=dashboard_info) 209 210 log_event( 211 "get_snapshot_data", 212 snapshot_type="report" if snapshot.is_report else "test_suite", 213 metrics=[m.get_id() for m in snapshot.first_level_metrics()], 214 metric_presets=snapshot.metadata.get(METRIC_PRESETS, []), 215 metric_generators=snapshot.metadata.get(METRIC_GENERATORS, []), 216 tests=[t.get_id() for t in snapshot.first_level_tests()], 217 test_presets=snapshot.metadata.get(TEST_PRESETS, []), 218 test_generators=snapshot.metadata.get(TEST_GENERATORS, []), 219 ) 220 return json.dumps(info.dict() if not isinstance(info, dict) else info, cls=NumpyEncoder) 221 222 223 @get("/{project_id:uuid}/{snapshot_id:uuid}/metadata") 224 async def get_snapshot_metadata( 225 snapshot_metadata: Annotated[SnapshotMetadata, Dependency()], 226 log_event: Callable, 227 ) -> SnapshotMetadata: 228 log_event( 229 "get_snapshot_metadata", 230 snapshot_type="report" if snapshot_metadata.is_report else "test_suite", 231 metric_presets=snapshot_metadata.metadata.get(METRIC_PRESETS, []), 232 metric_generators=snapshot_metadata.metadata.get(METRIC_GENERATORS, []), 233 test_presets=snapshot_metadata.metadata.get(TEST_PRESETS, []), 234 test_generators=snapshot_metadata.metadata.get(TEST_GENERATORS, []), 235 ) 236 return snapshot_metadata 237 238 239 @get("/{project_id:uuid}/dashboard/panels") 240 async def list_project_dashboard_panels( 241 project: Annotated[Project, Dependency()], 242 log_event: Callable, 243 ) -> List[DashboardPanel]: 244 log_event("list_project_dashboard_panels") 245 return list(project.dashboard.panels) 246 247 248 # We need this endpoint to export 249 # some additional models to open api schema 250 @get("/models/additional") 251 async def additional_models() -> ( 252 List[ 253 Union[ 254 DashboardInfoModel, 255 DashboardPanelPlot, 256 DashboardPanelCounter, 257 DashboardPanelDistribution, 258 DashboardPanelHistogram, 259 DashboardPanelTestSuite, 260 DashboardPanelTestSuiteCounter, 261 ] 262 ] 263 ): 264 return [] 265 266 267 @get("/{project_id:uuid}/dashboard") 268 async def project_dashboard( 269 project: Annotated[Project, Dependency()], 270 # TODO: no datetime, as it unable to validate '2023-07-09T02:03' 271 log_event: Callable, 272 timestamp_start: Optional[str] = None, 273 timestamp_end: Optional[str] = None, 274 ) -> str: 275 timestamp_start_ = datetime.datetime.fromisoformat(timestamp_start) if timestamp_start else None 276 timestamp_end_ = datetime.datetime.fromisoformat(timestamp_end) if timestamp_end else None 277 278 info = await DashboardInfoModel.from_project_with_time_range( 279 project, 280 timestamp_start=timestamp_start_, 281 timestamp_end=timestamp_end_, 282 ) 283 log_event("project_dashboard") 284 return json.dumps(info.dict() if not isinstance(info, dict) else info, cls=NumpyEncoder) 285 286 287 @post("/") 288 async def add_project( 289 data: Project, 290 project_manager: Annotated[ProjectManager, Dependency(skip_validation=True)], 291 log_event: Callable, 292 user_id: UserID, 293 team_id: Optional[TeamID] = None, 294 org_id: Optional[OrgID] = None, 295 ) -> Project: 296 if team_id is None and data.team_id is not None: 297 team_id = data.team_id 298 elif org_id is None and data.org_id is not None: 299 org_id = data.org_id 300 301 p = await project_manager.add_project(data, user_id, team_id, org_id) 302 log_event("add_project") 303 return p 304 305 306 @delete("/{project_id:uuid}") 307 async def delete_project( 308 project_id: Annotated[ProjectID, Parameter(title="id of project")], 309 project_manager: Annotated[ProjectManager, Dependency(skip_validation=True)], 310 log_event: Callable, 311 user_id: UserID, 312 ) -> None: 313 await project_manager.delete_project(user_id, project_id) 314 log_event("delete_project") 315 316 317 @post("/{project_id:uuid}/snapshots") 318 async def add_snapshot( 319 project: Annotated[Project, Dependency()], 320 parsed_json: Snapshot, 321 project_manager: Annotated[ProjectManager, Dependency(skip_validation=True)], 322 log_event: Callable, 323 user_id: UserID, 324 ) -> None: 325 await project_manager.add_snapshot(user_id, project.id, parsed_json) 326 log_event("add_snapshot") 327 328 329 @delete("/{project_id:uuid}/{snapshot_id:uuid}") 330 async def delete_snapshot( 331 project: Annotated[Project, Dependency()], 332 snapshot_id: Annotated[SnapshotID, Parameter(title="id of snapshot")], 333 project_manager: Annotated[ProjectManager, Dependency(skip_validation=True)], 334 log_event: Callable, 335 user_id: UserID, 336 ) -> None: 337 await project_manager.delete_snapshot(user_id, project.id, snapshot_id) 338 log_event("delete_snapshot") 339 340 341 def create_projects_api(guard: Callable) -> Router: 342 return Router( 343 "/projects", 344 route_handlers=[ 345 Router( 346 "", 347 route_handlers=[ 348 additional_models, 349 list_projects, 350 list_reports, 351 get_project_info, 352 search_projects, 353 list_test_suites, 354 get_snapshot_graph_data, 355 get_snapshot_data, 356 get_snapshot_download, 357 list_project_dashboard_panels, 358 project_dashboard, 359 list_snapshots, 360 get_snapshot_metadata, 361 ], 362 ), 363 Router( 364 "", 365 route_handlers=[ 366 update_project_info, 367 reload_project_snapshots, 368 add_project, 369 delete_project, 370 add_snapshot, 371 delete_snapshot, 372 ], 373 guards=[guard], 374 ), 375 ], 376 ) 377 378 379 projects_api_dependencies: Dict[str, Provide] = { 380 "project": Provide(path_project_dependency), 381 "snapshot_metadata": Provide(path_snapshot_metadata_dependency), 382 }