feat: updating serialization of functions

This commit is contained in:
Mose Müller 2023-08-10 15:48:45 +02:00
parent 06f93c110e
commit d94b85f305

View File

@ -3,7 +3,7 @@ import inspect
import json import json
import os import os
from enum import Enum from enum import Enum
from typing import Any, Optional, Union, cast, get_args, get_type_hints from typing import Any, Optional, cast, get_type_hints
import rpyc import rpyc
from loguru import logger from loguru import logger
@ -265,22 +265,13 @@ class DataService(rpyc.Service, AbstractDataService):
} }
elif inspect.isfunction(value) or inspect.ismethod(value): elif inspect.isfunction(value) or inspect.ismethod(value):
sig = inspect.signature(value) sig = inspect.signature(value)
# Store parameters and their anotations in a dictionary
parameters: dict[str, Optional[str]] = {} parameters: dict[str, Optional[str]] = {}
for k, v in sig.parameters.items(): for k, v in sig.parameters.items():
annotation = v.annotation annotation = v.annotation
if annotation is not inspect._empty: if annotation is not inspect._empty:
if ( if isinstance(annotation, type):
hasattr(annotation, "__origin__")
and annotation.__origin__ is Union
):
# Handle union types
union_types = get_args(annotation)
type_names = [
t.__name__ if isinstance(t, type) else str(t)
for t in union_types
]
parameters[k] = " | ".join(type_names)
elif isinstance(annotation, type):
# Handle regular types # Handle regular types
parameters[k] = annotation.__name__ parameters[k] = annotation.__name__
else: else: