Spaces:
Running
Running
File size: 5,153 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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 |
"""
Transformations service layer using API.
"""
from datetime import datetime
from typing import Any, Dict, List, Union
from loguru import logger
from api.client import api_client
from open_notebook.domain.transformation import Transformation
class TransformationsService:
"""Service layer for transformations operations using API."""
def __init__(self):
logger.info("Using API for transformations operations")
def get_all_transformations(self) -> List[Transformation]:
"""Get all transformations."""
transformations_data = api_client.get_transformations()
# Convert API response to Transformation objects
transformations = []
for trans_data in transformations_data:
transformation = Transformation(
name=trans_data["name"],
title=trans_data["title"],
description=trans_data["description"],
prompt=trans_data["prompt"],
apply_default=trans_data["apply_default"],
)
transformation.id = trans_data["id"]
transformation.created = datetime.fromisoformat(trans_data["created"].replace('Z', '+00:00'))
transformation.updated = datetime.fromisoformat(trans_data["updated"].replace('Z', '+00:00'))
transformations.append(transformation)
return transformations
def get_transformation(self, transformation_id: str) -> Transformation:
"""Get a specific transformation."""
response = api_client.get_transformation(transformation_id)
trans_data = response if isinstance(response, dict) else response[0]
transformation = Transformation(
name=trans_data["name"],
title=trans_data["title"],
description=trans_data["description"],
prompt=trans_data["prompt"],
apply_default=trans_data["apply_default"],
)
transformation.id = trans_data["id"]
transformation.created = datetime.fromisoformat(trans_data["created"].replace('Z', '+00:00'))
transformation.updated = datetime.fromisoformat(trans_data["updated"].replace('Z', '+00:00'))
return transformation
def create_transformation(
self,
name: str,
title: str,
description: str,
prompt: str,
apply_default: bool = False
) -> Transformation:
"""Create a new transformation."""
response = api_client.create_transformation(
name=name,
title=title,
description=description,
prompt=prompt,
apply_default=apply_default
)
trans_data = response if isinstance(response, dict) else response[0]
transformation = Transformation(
name=trans_data["name"],
title=trans_data["title"],
description=trans_data["description"],
prompt=trans_data["prompt"],
apply_default=trans_data["apply_default"],
)
transformation.id = trans_data["id"]
transformation.created = datetime.fromisoformat(trans_data["created"].replace('Z', '+00:00'))
transformation.updated = datetime.fromisoformat(trans_data["updated"].replace('Z', '+00:00'))
return transformation
def update_transformation(self, transformation: Transformation) -> Transformation:
"""Update a transformation."""
if not transformation.id:
raise ValueError("Transformation ID is required for update")
updates = {
"name": transformation.name,
"title": transformation.title,
"description": transformation.description,
"prompt": transformation.prompt,
"apply_default": transformation.apply_default,
}
response = api_client.update_transformation(transformation.id, **updates)
trans_data = response if isinstance(response, dict) else response[0]
# Update the transformation object with the response
transformation.name = trans_data["name"]
transformation.title = trans_data["title"]
transformation.description = trans_data["description"]
transformation.prompt = trans_data["prompt"]
transformation.apply_default = trans_data["apply_default"]
transformation.updated = datetime.fromisoformat(trans_data["updated"].replace('Z', '+00:00'))
return transformation
def delete_transformation(self, transformation_id: str) -> bool:
"""Delete a transformation."""
api_client.delete_transformation(transformation_id)
return True
def execute_transformation(
self,
transformation_id: str,
input_text: str,
model_id: str
) -> Union[Dict[Any, Any], List[Dict[Any, Any]]]:
"""Execute a transformation on input text."""
result = api_client.execute_transformation(
transformation_id=transformation_id,
input_text=input_text,
model_id=model_id
)
return result
# Global service instance
transformations_service = TransformationsService() |