[python] add lint check for src code in CI

packages/http-client-python/eng/scripts/Build-Packages.ps1

@@ -57,6 +57,9 @@ try {
 
     Invoke-LoggedCommand "npm run build" -GroupOutput
 
+    Write-Host "run lint check for pygen"
+    Invoke-LoggedCommand "npm run lint:py" -GroupOutput
+
     # pack the emitter
     Invoke-LoggedCommand "npm pack"
     Copy-Item "typespec-http-client-python-$emitterVersion.tgz" -Destination "$outputPath/packages"

packages/http-client-python/eng/scripts/ci/utils.ts

@@ -11,6 +11,8 @@ const argv = parseArgs({
   args: process.argv.slice(2),
   options: {
     pythonPath: { type: "string" },
+    folderName: { type: "string" },
+    command: { type: "string" },
   },
 });
 

packages/http-client-python/generator/pygen/black.py

@@ -51,6 +51,7 @@ def process(self) -> bool:
         return True
 
     def format_file(self, file: Path) -> None:
+        file_content = ""
         try:
             file_content = self.read_file(file)
             file_content = black.format_file_contents(file_content, fast=True, mode=_BLACK_MODE)

packages/http-client-python/generator/pygen/codegen/__init__.py

@@ -249,8 +249,15 @@ def sort_exceptions(yaml_data: Dict[str, Any]) -> None:
                     if not operation.get("exceptions"):
                         continue
                     # sort exceptions by status code, first single status code, then range, then default
-                    operation["exceptions"] = sorted(operation["exceptions"], key=lambda x: 3 if x["statusCodes"][0] == "default" else (1 if isinstance(x["statusCodes"][0], int) else 2))
-
+                    operation["exceptions"] = sorted(
+                        operation["exceptions"],
+                        key=lambda x: (
+                            3
+                            if x["statusCodes"][0] == "default"
+                            else (1 if isinstance(x["statusCodes"][0], int) else 2)
+                        ),
+                    )
+
     @staticmethod
     def remove_cloud_errors(yaml_data: Dict[str, Any]) -> None:
         for client in yaml_data["clients"]:
@@ -325,7 +332,7 @@ def process(self) -> bool:
         self._validate_code_model_options()
         options = self._build_code_model_options()
         yaml_data = self.get_yaml()
-        
+
         self.sort_exceptions(yaml_data)
 
         if self.options_retriever.azure_arm:

packages/http-client-python/generator/pygen/codegen/models/lro_paging_operation.py

@@ -3,7 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-from typing import Any
+from typing import Any, List, Union
 from .imports import FileImport
 from .lro_operation import LROOperationBase
 from .paging_operation import PagingOperationBase
@@ -12,7 +12,7 @@
 
 class LROPagingOperation(LROOperationBase[LROPagingResponse], PagingOperationBase[LROPagingResponse]):
     @property
-    def success_status_codes(self):
+    def success_status_codes(self) -> List[Union[int, str, List[int]]]:
         """The list of all successfull status code."""
         return [200]
 

packages/http-client-python/generator/pygen/codegen/models/operation.py

@@ -3,19 +3,16 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-from itertools import chain
 from typing import (
     Dict,
     List,
     Any,
     Optional,
-    Tuple,
     Union,
     TYPE_CHECKING,
     Generic,
     TypeVar,
     cast,
-    Sequence,
 )
 
 from .request_builder_parameter import RequestBuilderParameter
@@ -206,7 +203,9 @@ def default_error_deserialization(self) -> Optional[str]:
 
     @property
     def non_default_errors(self) -> List[Response]:
-        return [e for e in self.exceptions if "default" not in e.status_codes and e.type and isinstance(e.type, ModelType)]
+        return [
+            e for e in self.exceptions if "default" not in e.status_codes and e.type and isinstance(e.type, ModelType)
+        ]
 
     def _imports_shared(self, async_mode: bool, **kwargs: Any) -> FileImport:  # pylint: disable=unused-argument
         file_import = FileImport(self.code_model)
@@ -419,7 +418,9 @@ def imports(  # pylint: disable=too-many-branches, disable=too-many-statements
             elif any(r.type for r in self.responses):
                 file_import.add_submodule_import(f"{relative_path}_model_base", "_deserialize", ImportType.LOCAL)
             if self.default_error_deserialization or self.non_default_errors:
-                file_import.add_submodule_import(f"{relative_path}_model_base", "_failsafe_deserialize", ImportType.LOCAL)
+                file_import.add_submodule_import(
+                    f"{relative_path}_model_base", "_failsafe_deserialize", ImportType.LOCAL
+                )
         return file_import
 
     def get_response_from_status(self, status_code: Optional[Union[str, int]]) -> ResponseType:
@@ -429,7 +430,7 @@ def get_response_from_status(self, status_code: Optional[Union[str, int]]) -> Re
             raise ValueError(f"Incorrect status code {status_code}, operation {self.name}") from exc
 
     @property
-    def success_status_codes(self) -> Sequence[Union[str, int]]:
+    def success_status_codes(self) -> List[Union[int, str, List[int]]]:
         """The list of all successfull status code."""
         return sorted([code for response in self.responses for code in response.status_codes])
 

packages/http-client-python/generator/pygen/codegen/serializers/builder_serializer.py

@@ -991,9 +991,7 @@ def handle_error_response(self, builder: OperationType) -> List[str]:
         elif isinstance(builder.stream_value, str):  # _stream is not sure, so we need to judge it
             retval.append("    if _stream:")
             retval.extend([f"    {l}" for l in response_read])
-        retval.append(
-            f"    map_error(status_code=response.status_code, response=response, error_map=error_map)"
-        )
+        retval.append("    map_error(status_code=response.status_code, response=response, error_map=error_map)")
         error_model = ""
         if builder.non_default_errors and self.code_model.options["models_mode"]:
             error_model = ", model=error"
@@ -1005,10 +1003,10 @@ def handle_error_response(self, builder: OperationType) -> List[str]:
                     for status_code in e.status_codes:
                         retval.append(f"    {condition} response.status_code == {status_code}:")
                         if self.code_model.options["models_mode"] == "dpg":
-                            retval.append(f"        error = _failsafe_deserialize({e.type.type_annotation(is_operation_file=True, skip_quote=True)},  response.json())")
+                            retval.append(f"        error = _failsafe_deserialize({e.type.type_annotation(is_operation_file=True, skip_quote=True)},  response.json())")  # type: ignore # pylint: disable=line-too-long
                         else:
                             retval.append(
-                                f"        error = self._deserialize.failsafe_deserialize({e.type.type_annotation(is_operation_file=True, skip_quote=True)}, "
+                                f"        error = self._deserialize.failsafe_deserialize({e.type.type_annotation(is_operation_file=True, skip_quote=True)}, "  # type: ignore # pylint: disable=line-too-long
                                 "pipeline_response)"
                             )
                         # add build-in error type
@@ -1043,12 +1041,14 @@ def handle_error_response(self, builder: OperationType) -> List[str]:
                             )
                 # ranged status code only exist in typespec and will not have multiple status codes
                 else:
-                    retval.append(f"    {condition} {e.status_codes[0][0]} <= response.status_code <= {e.status_codes[0][1]}:")
+                    retval.append(
+                        f"    {condition} {e.status_codes[0][0]} <= response.status_code <= {e.status_codes[0][1]}:"
+                    )
                     if self.code_model.options["models_mode"] == "dpg":
-                        retval.append(f"        error = _failsafe_deserialize({e.type.type_annotation(is_operation_file=True, skip_quote=True)},  response.json())")
+                        retval.append(f"        error = _failsafe_deserialize({e.type.type_annotation(is_operation_file=True, skip_quote=True)},  response.json())")  # type: ignore  # pylint: disable=line-too-long
                     else:
                         retval.append(
-                            f"        error = self._deserialize.failsafe_deserialize({e.type.type_annotation(is_operation_file=True, skip_quote=True)}, "
+                            f"        error = self._deserialize.failsafe_deserialize({e.type.type_annotation(is_operation_file=True, skip_quote=True)}, "  # type: ignore  # pylint: disable=line-too-long
                             "pipeline_response)"
                         )
                 condition = "elif"
@@ -1059,7 +1059,9 @@ def handle_error_response(self, builder: OperationType) -> List[str]:
             if builder.non_default_errors:
                 retval.append("    else:")
             if self.code_model.options["models_mode"] == "dpg":
-                retval.append(f"{indent}error = _failsafe_deserialize({builder.default_error_deserialization},  response.json())")
+                retval.append(
+                    f"{indent}error = _failsafe_deserialize({builder.default_error_deserialization},  response.json())"
+                )
             else:
                 retval.append(
                     f"{indent}error = self._deserialize.failsafe_deserialize({builder.default_error_deserialization}, "
@@ -1086,7 +1088,7 @@ def handle_response(self, builder: OperationType) -> List[str]:
             if len(builder.responses) > 1:
                 status_codes, res_headers, res_deserialization = [], [], []
                 for status_code in builder.success_status_codes:
-                    response = builder.get_response_from_status(status_code)
+                    response = builder.get_response_from_status(status_code)  # type: ignore
                     if response.headers or response.type:
                         status_codes.append(status_code)
                         res_headers.append(self.response_headers(response))
@@ -1143,10 +1145,13 @@ def _need_specific_error_map(self, code: int, builder: OperationType) -> bool:
             if code in non_default_error.status_codes:
                 return False
             # ranged status code
-            if isinstance(non_default_error.status_codes[0], list) and non_default_error.status_codes[0][0] <= code <= non_default_error.status_codes[0][1]:
+            if (
+                isinstance(non_default_error.status_codes[0], list)
+                and non_default_error.status_codes[0][0] <= code <= non_default_error.status_codes[0][1]
+            ):
                 return False
         return True
-    
+
     def error_map(self, builder: OperationType) -> List[str]:
         retval = ["error_map: MutableMapping = {"]
         if builder.non_default_errors and self.code_model.options["models_mode"]:

packages/http-client-python/generator/test/azure/mock_api_tests/asynctests/test_azure_arm_resource_async.py

@@ -36,9 +36,7 @@ async def test_client_signature(credential, authentication_policy):
         # make sure signautre order is correct
         await client.top_level.get(RESOURCE_GROUP_NAME, "top")
         # make sure signautre name is correct
-        await client.top_level.get(
-            resource_group_name=RESOURCE_GROUP_NAME, top_level_tracked_resource_name="top"
-        )
+        await client.top_level.get(resource_group_name=RESOURCE_GROUP_NAME, top_level_tracked_resource_name="top")
 
 
 @pytest.mark.asyncio

packages/http-client-python/generator/test/azure/mock_api_tests/test_azure_arm_resource.py

@@ -35,9 +35,7 @@ def test_client_signature(credential, authentication_policy):
         # make sure signautre order is correct
         client.top_level.get(RESOURCE_GROUP_NAME, "top")
         # make sure signautre name is correct
-        client.top_level.get(
-            resource_group_name=RESOURCE_GROUP_NAME, top_level_tracked_resource_name="top"
-        )
+        client.top_level.get(resource_group_name=RESOURCE_GROUP_NAME, top_level_tracked_resource_name="top")
 
 
 def test_top_level_begin_create_or_replace(client):

packages/http-client-python/package.json

@@ -33,7 +33,7 @@
     "build": "tsc -p ./emitter/tsconfig.build.json",
     "watch": "tsc -p ./emitter/tsconfig.build.json --watch",
     "lint": "eslint . --max-warnings=0",
-    "lint:py": "tsx ./eng/scripts/ci/lint.ts --folderName generator",
+    "lint:py": "tsx ./eng/scripts/ci/lint.ts --folderName generator/pygen",
     "format": "pnpm -w format:dir packages/http-client-python && tsx ./eng/scripts/ci/format.ts",
     "install": "tsx ./eng/scripts/setup/run-python3.ts ./eng/scripts/setup/install.py",
     "prepare": "tsx ./eng/scripts/setup/run-python3.ts ./eng/scripts/setup/prepare.py",