andras/great-ai
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases 4 Packages Wiki Activity Actions

Add builtin deserialization

Browse source
This commit is contained in:
Andras Schmelczer 2022-07-01 20:34:47 +02:00
parent 8692cc60ee
commit 6a50f87abe
2 changed files with 15 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

12
src/great_ai/great_ai/remote/call_remote_great_ai.py
View file

@ -1,5 +1,7 @@
import asyncio
from typing import Any, Mapping
from typing import Any, Mapping, Optional, TypeVar
from pydantic import BaseModel
from ...utilities import get_logger
from ..views import Trace
@ -7,10 +9,10 @@ from .call_remote_great_ai_async import call_remote_great_ai_async
logger = get_logger("call_remote_great_ai")
T = TypeVar("T", BaseModel)
def call_remote_great_ai(
base_uri: str, data: Mapping[str, Any], retry_count: int = 4
) -> Trace:
base_uri: str, data: Mapping[str, Any], retry_count: int = 4, model_class:Optional[Type[T]]=None
) -> Trace[T]:
try:
asyncio.get_running_loop()
raise Exception(
@ -20,7 +22,7 @@ def call_remote_great_ai(
pass
future = call_remote_great_ai_async(
base_uri=base_uri, data=data, retry_count=retry_count
base_uri=base_uri, data=data, retry_count=retry_count,model_class=model_class
)
return asyncio.run(future)

11
src/great_ai/great_ai/remote/call_remote_great_ai_async.py
View file

@ -1,4 +1,6 @@
from typing import Any, Mapping, Optional
from typing import Any, Mapping, Optional, Type, TypeVar
from pydantic import BaseModel
from ..views import Trace
from .http_client import HttpClient
@ -6,10 +8,11 @@ from .remote_call_error import RemoteCallError
http: Optional[HttpClient] = None
T = TypeVar("T", BaseModel)
async def call_remote_great_ai_async(
base_uri: str, data: Mapping[str, Any], retry_count: int = 4
) -> Trace:
base_uri: str, data: Mapping[str, Any], retry_count: int = 4, model_class:Optional[Type[T]]=None
) -> Trace[T]:
global http
if http is None:
http = HttpClient()
@ -23,6 +26,8 @@ async def call_remote_great_ai_async(
)
try:
if model_class is not None:
response['output'] = model_class.parse_obj(response['output'])
return Trace.parse_obj(response)
except Exception:
raise RemoteCallError("Could not parse response")