Merge pull request #224 from galaxyproject/fix_model_conversion

Simplify and fix model conversion

server/galaxyls/config.py

@@ -15,33 +15,33 @@ class CompletionConfig:
     """Auto-completion feature configuration."""
 
     mode: CompletionMode = attrs.field(default=CompletionMode.AUTO)
-    auto_close_tags: bool = attrs.field(default=True, alias="autoCloseTags")
+    auto_close_tags: bool = attrs.field(default=True)
 
 
 @attrs.define
 class ServerConfig:
     """Language Server specific configuration."""
 
-    silent_install: bool = attrs.field(default=False, alias="silentInstall")
+    silent_install: bool = attrs.field(default=False)
 
 
 @attrs.define
 class PlanemoTestingConfig:
     """Planemo testing configuration."""
 
     enabled: bool = attrs.field(default=True)
-    auto_test_discover_on_save_enabled: bool = attrs.field(default=True, alias="autoTestDiscoverOnSaveEnabled")
-    extra_params: str = attrs.field(default="", alias="extraParams")
+    auto_test_discover_on_save_enabled: bool = attrs.field(default=True)
+    extra_params: str = attrs.field(default="")
 
 
 @attrs.define
 class PlanemoConfig:
     """Planemo integration configuration."""
 
     enabled: bool = attrs.field(default=False)
-    env_path: str = attrs.field(default="planemo", alias="envPath")
-    galaxy_root: Optional[str] = attrs.field(default=None, alias="galaxyRoot")
-    get_cwd: Optional[str] = attrs.field(default=None, alias="getCwd")
+    env_path: str = attrs.field(default="planemo")
+    galaxy_root: Optional[str] = attrs.field(default=None)
+    get_cwd: Optional[str] = attrs.field(default=None)
     testing: PlanemoTestingConfig = attrs.field(default=PlanemoTestingConfig())
 
 
@@ -52,13 +52,3 @@ class GalaxyToolsConfiguration:
     server: ServerConfig = attrs.field(default=ServerConfig())
     completion: CompletionConfig = attrs.field(default=CompletionConfig())
     planemo: PlanemoConfig = attrs.field(default=PlanemoConfig())
-
-    @classmethod
-    def from_config_dict(cls, config: dict):
-        result = GalaxyToolsConfiguration(
-            server=ServerConfig(**config["server"]),
-            completion=CompletionConfig(**config["completion"]),
-            planemo=PlanemoConfig(**config["planemo"]),
-        )
-        result.planemo.testing = PlanemoTestingConfig(**config["planemo"]["testing"])
-        return result

server/galaxyls/server.py

@@ -59,7 +59,7 @@
     ReplaceTextRangeResult,
     TestSuiteInfoResult,
 )
-from galaxyls.utils import deserialize_command_param
+from galaxyls.utils import convert_to
 
 GLS_VERSION = "0.10.1"
 GLS_NAME = "galaxy-tools-language-server"
@@ -88,7 +88,7 @@ async def _load_client_config_async(server: GalaxyToolsLanguageServer) -> None:
         config = await server.get_configuration_async(
             WorkspaceConfigurationParams(items=[ConfigurationItem(section="galaxyTools")])
         )
-        server.configuration = GalaxyToolsConfiguration.from_config_dict(config[0])
+        server.configuration = convert_to(config[0], GalaxyToolsConfiguration)
     except BaseException as err:
         server.show_message_log(f"Error loading configuration: {err}")
         server.show_message("Error loading configuration. Using default settings.", MessageType.Error)
@@ -190,7 +190,7 @@ def process_code_actions(server: GalaxyToolsLanguageServer, params: CodeActionPa
 def auto_close_tag(server: GalaxyToolsLanguageServer, parameters: CommandParameters) -> Optional[AutoCloseTagResult]:
     """Responds to a close tag request to close the currently opened node."""
     if server.configuration.completion.auto_close_tags and parameters:
-        params = deserialize_command_param(parameters[0], TextDocumentPositionParams)
+        params = convert_to(parameters[0], TextDocumentPositionParams)
         document = _get_valid_document(server, params.text_document.uri)
         if document:
             xml_document = _get_xml_document(document)
@@ -203,7 +203,7 @@ async def cmd_generate_test(
     server: GalaxyToolsLanguageServer, parameters: CommandParameters
 ) -> Optional[GeneratedSnippetResult]:
     """Generates some test snippets based on the inputs and outputs of the document."""
-    params = deserialize_command_param(parameters[0], TextDocumentIdentifier)
+    params = convert_to(parameters[0], TextDocumentIdentifier)
     document = _get_valid_document(server, params.uri)
     if document:
         return server.service.generate_tests(document)
@@ -215,7 +215,7 @@ async def cmd_generate_command(
     server: GalaxyToolsLanguageServer, parameters: CommandParameters
 ) -> Optional[GeneratedSnippetResult]:
     """Generates a boilerplate Cheetah code snippet based on the inputs and outputs of the document."""
-    params = deserialize_command_param(parameters[0], TextDocumentIdentifier)
+    params = convert_to(parameters[0], TextDocumentIdentifier)
     document = _get_valid_document(server, params.uri)
     if document:
         return server.service.generate_command(document)
@@ -227,7 +227,7 @@ def sort_single_param_attrs_command(
     server: GalaxyToolsLanguageServer, parameters: CommandParameters
 ) -> Optional[ReplaceTextRangeResult]:
     """Sorts the attributes of the param element under the cursor."""
-    params = deserialize_command_param(parameters[0], TextDocumentPositionParams)
+    params = convert_to(parameters[0], TextDocumentPositionParams)
     document = _get_valid_document(server, params.text_document.uri)
     if document:
         xml_document = _get_xml_document(document)
@@ -240,7 +240,7 @@ def sort_document_params_attrs_command(
     server: GalaxyToolsLanguageServer, parameters: CommandParameters
 ) -> Optional[List[ReplaceTextRangeResult]]:
     """Sorts the attributes of all the param elements contained in the document."""
-    params = deserialize_command_param(parameters[0], TextDocumentIdentifier)
+    params = convert_to(parameters[0], TextDocumentIdentifier)
     document = _get_valid_document(server, params.uri)
     if document:
         xml_document = _get_xml_document(document)
@@ -251,7 +251,7 @@ def sort_document_params_attrs_command(
 @language_server.command(Commands.GENERATE_EXPANDED_DOCUMENT)
 def generate_expanded_command(server: GalaxyToolsLanguageServer, parameters: CommandParameters) -> GeneratedExpandedDocument:
     """Generates a expanded version (with all macros replaced) of the tool document."""
-    params = deserialize_command_param(parameters[0], TextDocumentIdentifier)
+    params = convert_to(parameters[0], TextDocumentIdentifier)
     document = server.workspace.get_document(params.uri)
     if document and DocumentValidator.is_tool_document(document):
         return server.service.macro_expander.generate_expanded_from(document.path)
@@ -271,7 +271,7 @@ def discover_tests_in_document_command(
     server: GalaxyToolsLanguageServer, parameters: CommandParameters
 ) -> Optional[TestSuiteInfoResult]:
     """Returns a test suite containing all tests for a particular XML tool document."""
-    params = deserialize_command_param(parameters[0], TextDocumentIdentifier)
+    params = convert_to(parameters[0], TextDocumentIdentifier)
     document = _get_valid_document(server, params.uri)
     if document:
         xml_document = _get_xml_document(document)

server/galaxyls/tests/unit/test_config.py

@@ -4,6 +4,7 @@
     GalaxyToolsConfiguration,
     ServerConfig,
 )
+from galaxyls.utils import convert_to
 
 
 class TestGalaxyToolsConfigurationClass:
@@ -32,7 +33,7 @@ def test_init_configuration_from_dict(self):
             },
         }
 
-        config = GalaxyToolsConfiguration.from_config_dict(config_dict)
+        config = convert_to(config_dict, GalaxyToolsConfiguration)
 
         assert config.server.silent_install is False
         assert config.completion.mode == CompletionMode.DISABLED

server/galaxyls/utils.py

@@ -4,37 +4,13 @@
     TypeVar,
 )
 
-T = TypeVar("T")
-
-
-def is_namedtuple_instance(x) -> bool:
-    _type = type(x)
-    bases = _type.__bases__
-    if len(bases) != 1 or bases[0] != tuple:
-        return False
-    fields = getattr(_type, "_fields", None)
-    if not isinstance(fields, tuple):
-        return False
-    return all(type(i) == str for i in fields)
+from lsprotocol import converters as cv
 
-
-def unpack(obj):
-    """Unpacks a named tuple into a dictionary.
-    https://stackoverflow.com/a/39235373"""
-    if isinstance(obj, dict):
-        return {key: unpack(value) for key, value in obj.items()}
-    elif isinstance(obj, list):
-        return [unpack(value) for value in obj]
-    elif is_namedtuple_instance(obj):
-        return {key: unpack(value) for key, value in obj._asdict().items()}
-    elif isinstance(obj, tuple):
-        return tuple(unpack(value) for value in obj)
-    else:
-        return obj
+T = TypeVar("T")
 
 
-def deserialize_command_param(params: NamedTuple, type: Type[T]) -> T:
+def convert_to(params: NamedTuple, type: Type[T]) -> T:
     """Given a namedtuple, converts it into the given model type."""
-    params_dict = unpack(params)
-    obj = type(**params_dict)  # type: ignore [call-arg]
+    converter = cv.get_converter()
+    obj = converter.structure(params, type)
     return obj