LRPC-9 Use PyLint

* Disable pylint line-too-long-check. Already taken care of by black
* Added function to load LRPC schema
* LRPC requires at least Python 3.9
* More type hints
* Some refactoring

package/lrpc/PlantUmlVisitor.py

@@ -1,3 +1,4 @@
+import os
 from typing import Optional, Union
 from contextlib import contextmanager
 
@@ -166,7 +167,8 @@ def function_string(self, fsb: FunctionStringBuilder) -> None:
 
 
 class PlantUmlVisitor(LrpcVisitor):
-    def __init__(self, output: str) -> None:
+
+    def __init__(self, output: os.PathLike) -> None:
         self.output = output
         self.fsb: FunctionStringBuilder
         self.puml: PumlFile

package/lrpc/client/client_cli_visitor.py

@@ -1,4 +1,5 @@
 from functools import partial
+from typing import Any, Callable, Optional
 
 import click
 import yaml
@@ -16,7 +17,7 @@ class YamlParamType(click.ParamType):
 
     name = "YAML"
 
-    def convert(self, value, param, ctx):
+    def convert(self, value: Any, param: Optional[click.Parameter], ctx: Optional[click.Context]) -> Any:
         if isinstance(value, dict):
             return value
 
@@ -37,7 +38,7 @@ class OptionalParamType(click.ParamType):
     def __init__(self, contained_type: click.ParamType) -> None:
         self.contained_type: click.ParamType = contained_type
 
-    def convert(self, value, param, ctx):
+    def convert(self, value: Any, param: Optional[click.Parameter], ctx: Optional[click.Context]) -> Any:
         if not isinstance(value, str):
             self.fail(f"{value} is not a string", param, ctx)
 
@@ -56,43 +57,43 @@ class ClientCliVisitor(LrpcVisitor):
     Call ClientCliVisitor.root() to activate the CLI
     """
 
-    def __init__(self, callback) -> None:
-        self.root: click.Group = None
-        self.current_service: click.Group = None
-        self.current_function: click.Command = None
+    def __init__(self, callback: Callable) -> None:
+        self.root: click.Group
+        self.current_service: click.Group
+        self.current_function: click.Command
         self.enum_values: dict[str, list[str]] = {}
 
         self.callback = callback
 
-    def visit_lrpc_def(self, lrpc_def: LrpcDef):
+    def visit_lrpc_def(self, lrpc_def: LrpcDef) -> None:
         self.root = click.Group(lrpc_def.name())
 
-    def visit_lrpc_service(self, service: LrpcService):
+    def visit_lrpc_service(self, service: LrpcService) -> None:
         self.current_service = click.Group(service.name())
 
-    def visit_lrpc_service_end(self):
+    def visit_lrpc_service_end(self) -> None:
         self.root.add_command(self.current_service)
 
-    def visit_lrpc_enum_field(self, enum: LrpcEnum, field: LrpcEnumField):
+    def visit_lrpc_enum_field(self, enum: LrpcEnum, field: LrpcEnumField) -> None:
         if enum.name() not in self.enum_values:
             self.enum_values.update({enum.name(): []})
 
         self.enum_values[enum.name()].append(field.name())
 
-    def visit_lrpc_function(self, function: LrpcFun):
+    def visit_lrpc_function(self, function: LrpcFun) -> None:
         self.current_function = click.Command(
             name=function.name(),
             callback=partial(self.__handle_command, self.current_service.name, function.name()),
             help="my help 123",
         )
 
-    def visit_lrpc_function_end(self):
+    def visit_lrpc_function_end(self) -> None:
         self.current_service.add_command(self.current_function)
 
-    def visit_lrpc_function_param(self, param: LrpcVar):
+    def visit_lrpc_function_param(self, param: LrpcVar) -> None:
         attributes = {"type": self.__click_type(param), "nargs": param.array_size() if param.is_array() else 1}
 
-        arg = click.Argument([param.name()], **attributes)
+        arg = click.Argument([param.name()], True, **attributes)
         self.current_function.params.append(arg)
 
     def __click_type(self, param: LrpcVar) -> click.ParamType:
@@ -121,7 +122,7 @@ def __click_type(self, param: LrpcVar) -> click.ParamType:
         else:
             return t
 
-    def __handle_command(self, service, function, **kwargs):
+    def __handle_command(self, service: Optional[str], function: str, **kwargs: Any) -> Callable:
         for a, v in kwargs.items():
             if v == NONE_ARG:
                 kwargs[a] = None

package/lrpc/client/lrpc_client.py

@@ -1,15 +1,16 @@
 import struct
+from typing import Optional, Any
 
+from lrpc.client import LrpcDecoder, lrpc_encode
 from lrpc.core import LrpcDef
-from lrpc.client import lrpc_encode, LrpcDecoder
 
 
 class LrpcClient:
     def __init__(self, definition_file: str) -> None:
         self.lrpc_def: LrpcDef = LrpcDef.load(definition_file)
         self.receive_buffer = b""
 
-    def process(self, encoded: bytes):
+    def process(self, encoded: bytes) -> Optional[dict[str, Any]]:
         self.receive_buffer += encoded
         received = len(self.receive_buffer)
 
@@ -25,7 +26,7 @@ def process(self, encoded: bytes):
 
         return None
 
-    def decode(self, encoded: bytes):
+    def decode(self, encoded: bytes) -> dict[str, Any]:
         ret = {}
 
         # skip packet length at index 0
@@ -46,7 +47,7 @@ def decode(self, encoded: bytes):
 
         return ret
 
-    def encode(self, service_name: str, function_name: str, **kwargs):
+    def encode(self, service_name: str, function_name: str, **kwargs: Any) -> bytes:
         service = self.lrpc_def.service_by_name(service_name)
         if not service:
             raise ValueError(f"Service {service_name} not found in the LRPC definition file")

package/lrpc/codegen/constants.py

@@ -1,3 +1,4 @@
+import os
 from typing import Optional
 from code_generation.code_generator import CppFile
 from lrpc import LrpcVisitor
@@ -6,7 +7,8 @@
 
 
 class ConstantsFileVisitor(LrpcVisitor):
-    def __init__(self, output: str):
+
+    def __init__(self, output: os.PathLike):
         self.output = output
         self.namespace: Optional[str] = None
         self.file: CppFile

package/lrpc/codegen/enum.py

@@ -1,3 +1,4 @@
+import os
 from typing import Optional
 from code_generation.code_generator import CppFile
 from lrpc import LrpcVisitor
@@ -7,7 +8,8 @@
 
 
 class EnumFileVisitor(LrpcVisitor):
-    def __init__(self, output: str) -> None:
+
+    def __init__(self, output: os.PathLike) -> None:
         self.descriptor: LrpcEnum
         self.file: CppFile
         self.namespace: Optional[str]

package/lrpc/codegen/server_include.py

@@ -2,10 +2,11 @@
 from lrpc.core import LrpcDef, LrpcService
 from lrpc import LrpcVisitor
 from lrpc.codegen.utils import optionally_in_namespace
+import os
 
 
 class ServerIncludeVisitor(LrpcVisitor):
-    def __init__(self, output: str) -> None:
+    def __init__(self, output: os.PathLike) -> None:
         self.lrpc_def: LrpcDef
         self.output = output
         self.file: CppFile

package/lrpc/codegen/service_include.py

@@ -1,11 +1,13 @@
+import os
+
 from code_generation.code_generator import CppFile
-from lrpc.core import LrpcFun, LrpcService, LrpcVar
 from lrpc import LrpcVisitor
 from lrpc.codegen.common import lrpc_var_includes
+from lrpc.core import LrpcFun, LrpcService, LrpcVar
 
 
 class ServiceIncludeVisitor(LrpcVisitor):
-    def __init__(self, output: str) -> None:
+    def __init__(self, output: os.PathLike) -> None:
         self.output = output
         self.file: CppFile
         self.includes: set[str] = set()

package/lrpc/codegen/service_shim.py

@@ -1,11 +1,13 @@
+import os
 from code_generation.code_generator import CppFile
 from lrpc import LrpcVisitor
 from lrpc.core import LrpcDef, LrpcService, LrpcFun, LrpcVar
 from lrpc.codegen.utils import optionally_in_namespace
 
 
 class ServiceShimVisitor(LrpcVisitor):
-    def __init__(self, output: str):
+
+    def __init__(self, output: os.PathLike):
         self.file: CppFile
         self.lrpc_def: LrpcDef
         self.output = output

package/lrpc/codegen/struct.py

@@ -1,3 +1,4 @@
+import os
 from typing import Set
 
 from code_generation.code_generator import CppFile
@@ -8,7 +9,8 @@
 
 
 class StructFileVisitor(LrpcVisitor):
-    def __init__(self, output: str) -> None:
+
+    def __init__(self, output: os.PathLike) -> None:
         self.output = output
         self.namespace = None
         self.file: CppFile

package/lrpc/core/definition.py

@@ -1,11 +1,10 @@
-from importlib import resources
 from typing import Optional, TypedDict
 from typing_extensions import NotRequired
 
 import jsonschema
 import yaml
 from lrpc import LrpcVisitor
-from lrpc import schema as lrpc_schema
+from lrpc.schema import load_lrpc_schema
 from lrpc.core import LrpcConstant, LrpcEnum, LrpcService, LrpcStruct, LrpcVarDict
 from lrpc.core import LrpcStructDict, LrpcServiceDict, LrpcConstantDict, LrpcEnumDict
 
@@ -166,12 +165,9 @@ def __update_type(cls, var: LrpcVarDict, struct_names: list[str], enum_names: li
     def load(definition_url: str) -> "LrpcDef":
         from lrpc.validation import SemanticAnalyzer
 
-        schema_url = resources.files(lrpc_schema).joinpath("lotusrpc-schema.json")
-
         with open(definition_url, mode="rt", encoding="utf-8") as rpc_def:
             definition = yaml.safe_load(rpc_def)
-            schema = yaml.safe_load(schema_url.read_text())
-            jsonschema.validate(definition, schema)
+            jsonschema.validate(definition, load_lrpc_schema())
 
             lrpc_def = LrpcDef(definition)
             sa = SemanticAnalyzer(lrpc_def)

package/lrpc/core/function.py

@@ -1,4 +1,4 @@
-from typing import Optional, TypedDict
+from typing import TypedDict
 from typing_extensions import NotRequired
 
 from lrpc import LrpcVisitor
@@ -46,22 +46,22 @@ def accept(self, visitor: LrpcVisitor) -> None:
     def params(self) -> list[LrpcVar]:
         return self.__params
 
-    def param(self, name: str) -> Optional[LrpcVar]:
+    def param(self, name: str) -> LrpcVar:
         for p in self.params():
             if p.name() == name:
                 return p
 
-        return None
+        raise ValueError(f"No parameter {name} in function {self.name()}")
 
     def returns(self) -> list[LrpcVar]:
         return self.__returns
 
-    def ret(self, name: str) -> Optional[LrpcVar]:
+    def ret(self, name: str) -> LrpcVar:
         for r in self.returns():
             if r.name() == name:
                 return r
 
-        return None
+        raise ValueError(f"No return value {name} in function {self.name()}")
 
     def name(self) -> str:
         return self.__name

package/lrpc/lotusrpc.py

@@ -1,13 +1,13 @@
 import os
-from importlib import resources
 from os import path
+from typing import TextIO, Any
 
 import click
 import jsonschema
 import jsonschema.exceptions
 import yaml
 from lrpc import PlantUmlVisitor
-from lrpc import schema as lrpc_schema
+from lrpc.schema import load_lrpc_schema
 from lrpc.codegen import (
     ConstantsFileVisitor,
     EnumFileVisitor,
@@ -20,26 +20,22 @@
 from lrpc.validation import SemanticAnalyzer
 
 
-def create_dir_if_not_exists(target_dir) -> None:
+def create_dir_if_not_exists(target_dir: os.PathLike) -> None:
     if not path.exists(target_dir):
         os.makedirs(target_dir, 511, True)
 
 
-def validate_yaml(definition, input: str) -> bool:
-    url = resources.files(lrpc_schema) / "lotusrpc-schema.json"
-    with open(url, mode="rt", encoding="utf-8") as schema_file:
-        schema = yaml.safe_load(schema_file)
-
-        try:
-            jsonschema.validate(definition, schema)
-            return False
-        except jsonschema.exceptions.ValidationError as e:
-            print("#" * 80)
-            print(" LRPC definition parsing error ".center(80, "#"))
-            print(f" {input} ".center(80, "#"))
-            print("#" * 80)
-            print(e)
-            return True
+def validate_yaml(definition: dict[str, Any], input_file_name: str) -> bool:
+    try:
+        jsonschema.validate(definition, load_lrpc_schema())
+        return False
+    except jsonschema.exceptions.ValidationError as e:
+        print("#" * 80)
+        print(" LRPC definition parsing error ".center(80, "#"))
+        print(f" {input_file_name} ".center(80, "#"))
+        print("#" * 80)
+        print(e)
+        return True
 
 
 def validate_definition(lrpc_def: LrpcDef, warnings_as_errors: bool) -> bool:
@@ -59,7 +55,7 @@ def validate_definition(lrpc_def: LrpcDef, warnings_as_errors: bool) -> bool:
     return errors_found
 
 
-def generate_rpc(lrpc_def: LrpcDef, output: str) -> None:
+def generate_rpc(lrpc_def: LrpcDef, output: os.PathLike) -> None:
     create_dir_if_not_exists(output)
 
     lrpc_def.accept(ServerIncludeVisitor(output))
@@ -76,21 +72,23 @@ def generate_rpc(lrpc_def: LrpcDef, output: str) -> None:
     "-w", "--warnings_as_errors", help="Treat warnings as errors", required=False, default=None, is_flag=True, type=str
 )
 @click.option("-o", "--output", help="Path to put the generated files", required=False, default=".", type=click.Path())
-@click.argument("input", type=click.File("r"))
-def generate(warnings_as_errors: bool, output, input) -> None:
+@click.argument("input", "input_file", type=click.File("r"))
+def generate(warnings_as_errors: bool, output: os.PathLike, input_file: TextIO) -> None:
     """Generate code for file(s) INPUTS"""
 
-    definition = yaml.safe_load(input)
-    errors_found = validate_yaml(definition, input)
+    definition = yaml.safe_load(input_file)
+    errors_found = validate_yaml(definition, input_file.name)
     if errors_found:
         return
 
     lrpc_def = LrpcDef(definition)
     errors_found = validate_definition(lrpc_def, warnings_as_errors)
     if not errors_found:
-        input.seek(0)
+        input_file.seek(0)
         generate_rpc(lrpc_def, output)
 
 
 if __name__ == "__main__":
+    # parameters are inserted by Click
+    # pylint: disable=no-value-for-parameter
     generate()

package/lrpc/schema/__init__.py

@@ -0,0 +1 @@
+from .load import load_lrpc_schema

package/lrpc/schema/load.py

@@ -0,0 +1,11 @@
+from importlib import resources
+from typing import Any
+
+import yaml
+import lrpc.schema as lrpc_schema
+
+
+def load_lrpc_schema() -> Any:
+    schema_file = resources.files(lrpc_schema).joinpath("lotusrpc-schema.json")
+    schema_text = schema_file.read_text()
+    return yaml.safe_load(schema_text)