add type hints for python helpers

source/codegen/common_helpers.py

@@ -1204,48 +1204,48 @@ def get_params_needing_initialization(parameters: List[dict]) -> List[dict]:
     return [p for p in parameters if not (is_return_value(p) or is_get_last_error_output_param(p))]
 
 
-def filter_moniker_streaming_functions(functions, functions_to_generate):
+def filter_moniker_streaming_functions(functions: dict, functions_to_generate: List[str]) -> List[str]:
     """Return streaming functions that need to be generated."""
     return [
         name for name in functions_to_generate if is_moniker_streaming_function(functions[name])
     ]
 
 
-def is_moniker_streaming_function(function):
+def is_moniker_streaming_function(function: dict) -> bool:
     """Whether this function is for streaming data through moniker."""
     return function.get("is_streaming_api", False)
 
 
-def get_data_moniker_function_name(function_name):
+def get_data_moniker_function_name(function_name: str) -> str:
     """Return the corresponding moniker function name for the given C API function."""
     return function_name.replace("Begin", "Moniker")
 
 
-def get_data_moniker_struct_name(begin_function_name):
+def get_data_moniker_struct_name(begin_function_name: str) -> str:
     """Return the Moniker function name.
 
     Input expected is Begin* streaming API name.
     """
     return f"{begin_function_name.replace('Begin', 'Moniker')}Data"
 
 
-def get_data_moniker_request_message_type(begin_function_name):
+def get_data_moniker_request_message_type(begin_function_name: str) -> str:
     """Return the request message type for Moniker functions.
 
     Input expected is Begin* streaming API name.
     """
     return f"{begin_function_name.replace('Begin', '')}StreamingRequest"
 
 
-def get_data_moniker_response_message_type(begin_function_name):
+def get_data_moniker_response_message_type(begin_function_name: str) -> str:
     """Return the response message type for Moniker functions.
 
     Input expected is Begin* streaming API name.
     """
     return f"{begin_function_name.replace('Begin', '')}StreamingResponse"
 
 
-def get_data_moniker_function_parameters(function):
+def get_data_moniker_function_parameters(function: dict) -> tuple[List[dict], List[dict]]:
     """Return moniker function parameters split into input/output.
 
     Input expected is equivalent non-streaming function.
@@ -1261,12 +1261,12 @@ def get_data_moniker_function_parameters(function):
     return (input_parameters, output_parameters)
 
 
-def is_function_in_streaming_functions(function_name, streaming_functions_to_generate):
+def is_function_in_streaming_functions(function_name: str, streaming_functions_to_generate: List[str]):
     """Check if a function name is in the streaming functions to generate."""
     return function_name in streaming_functions_to_generate
 
 
-def _is_streaming_param_input_array(streaming_param):
+def _is_streaming_param_input_array(streaming_param: dict) -> bool:
     """Check if the streaming parameter is an input array."""
     return (
         streaming_param
@@ -1275,7 +1275,7 @@ def _is_streaming_param_input_array(streaming_param):
     )
 
 
-def get_non_streaming_input_parameters(parameters):
+def get_non_streaming_input_parameters(parameters: List[dict]) -> List[dict]:
     """Determine if a parameter should be passed from Begin streaming API to Moniker function."""
     streaming_param = get_first_streaming_parameter(parameters)
     params = []

source/codegen/service_helpers.py

@@ -709,15 +709,7 @@ def get_protobuf_cpplib_type(grpc_type: str) -> str:
     return grpc_type
 
 
-def get_streaming_type(parameters) -> str:
-    """Get the streaming type from the function data."""
-    for param in parameters:
-        if param.get("is_streaming_type", False):
-            return param["type"]
-    return None
-
-
-def get_size_param_name(streaming_param) -> str:
+def get_size_param_name(streaming_param: dict) -> str:
     """Get the size parameter name for the given streaming parameter.
 
     The size is only present for read arrays, which have an "out" direction.
@@ -726,11 +718,3 @@ def get_size_param_name(streaming_param) -> str:
         return common_helpers.camel_to_snake(streaming_param["size"]["value"])
     else:
         return None
-
-
-def get_c_api_name(function_name) -> str:
-    """Get the C API name for the given function name."""
-    if function_name.startswith("Begin"):
-        base_name = function_name[len("Begin") :]
-        return f"{base_name}"
-    return f"{function_name}"

source/codegen/templates/service_helpers.mako

@@ -313,7 +313,6 @@ ${populate_response(function_data=function_data, parameters=parameters)}\
   non_streaming_function_name = function_name.replace("Begin", "")
   non_streaming_function_parameters = functions[non_streaming_function_name]['parameters']
   arg_string = service_helpers.create_args(non_streaming_function_parameters)
-  c_api_name = service_helpers.get_c_api_name(function_name)
   output_parameters = [p for p in non_streaming_function_parameters if common_helpers.is_output_parameter(p)]
 %>\
 ::grpc::Status ${moniker_function_name}(void* data, google::protobuf::Arena& arena, google::protobuf::Any& packedData)
@@ -327,7 +326,7 @@ ${initialize_output_params(output_parameters)}\
 ${streaming_handle_in_direction(streaming_param)}\
     % endif
 
-    auto status = library->${c_api_name}(${arg_string});
+    auto status = library->${non_streaming_function_name}(${arg_string});
 
 ${populate_moniker_response_for_out_functions(output_parameters, streaming_param)}\
     return ::grpc::Status::OK;