Spaces:
Sleeping
Sleeping
File size: 3,402 Bytes
f871fed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | """
Insights service layer using API.
"""
from typing import List, Optional
from loguru import logger
from api.client import api_client
from open_notebook.domain.notebook import Note, SourceInsight
class InsightsService:
"""Service layer for insights operations using API."""
def __init__(self):
logger.info("Using API for insights operations")
def get_source_insights(self, source_id: str) -> List[SourceInsight]:
"""Get all insights for a specific source."""
insights_data = api_client.get_source_insights(source_id)
# Convert API response to SourceInsight objects
insights = []
for insight_data in insights_data:
insight = SourceInsight(
insight_type=insight_data["insight_type"],
content=insight_data["content"],
)
insight.id = insight_data["id"]
insight.created = insight_data["created"]
insight.updated = insight_data["updated"]
insights.append(insight)
return insights
def get_insight(self, insight_id: str) -> SourceInsight:
"""Get a specific insight."""
insight_response = api_client.get_insight(insight_id)
insight_data = insight_response if isinstance(insight_response, dict) else insight_response[0]
insight = SourceInsight(
insight_type=insight_data["insight_type"],
content=insight_data["content"],
)
insight.id = insight_data["id"]
insight.created = insight_data["created"]
insight.updated = insight_data["updated"]
# Note: source_id from API response is not stored; use await insight.get_source() if needed
return insight
def delete_insight(self, insight_id: str) -> bool:
"""Delete a specific insight."""
api_client.delete_insight(insight_id)
return True
def save_insight_as_note(self, insight_id: str, notebook_id: Optional[str] = None) -> Note:
"""Convert an insight to a note."""
note_response = api_client.save_insight_as_note(insight_id, notebook_id)
note_data = note_response if isinstance(note_response, dict) else note_response[0]
note = Note(
title=note_data["title"],
content=note_data["content"],
note_type=note_data["note_type"],
)
note.id = note_data["id"]
note.created = note_data["created"]
note.updated = note_data["updated"]
return note
def create_source_insight(self, source_id: str, transformation_id: str, model_id: Optional[str] = None) -> SourceInsight:
"""Create a new insight for a source by running a transformation."""
insight_response = api_client.create_source_insight(source_id, transformation_id, model_id)
insight_data = insight_response if isinstance(insight_response, dict) else insight_response[0]
insight = SourceInsight(
insight_type=insight_data["insight_type"],
content=insight_data["content"],
)
insight.id = insight_data["id"]
insight.created = insight_data["created"]
insight.updated = insight_data["updated"]
# Note: source_id from API response is not stored; use await insight.get_source() if needed
return insight
# Global service instance
insights_service = InsightsService() |