[python] add generation-subdir flag

.chronus/changes/python-addSubdir-2025-6-28-15-21-55.md

@@ -0,0 +1,7 @@
+---
+changeKind: feature
+packages:
+  - "@typespec/http-client-python"
+---
+
+add `generation-subdir` flag

packages/http-client-python/emitter/src/lib.ts

@@ -87,6 +87,12 @@ export const PythonEmitterOptionsSchema: JSONSchemaType<PythonEmitterOptions> =
       description:
         "Whether to validate the versioning of the package. Defaults to `true`. If set to `false`, we will not validate the versioning of the package.",
     },
+    "generation-subdir": {
+      type: "string",
+      nullable: true,
+      description:
+        "The subdirectory to generate the code in. If not specified, the code will be generated in the root folder. Note: if you're using this flag, you will need to add and maintain the versioning file yourself.",
+    },
     "keep-setup-py": {
       type: "boolean",
       nullable: true,

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

@@ -40,6 +40,7 @@ class OptionsDict(MutableMapping):
         "polymorphic-examples": 5,
         "validate-versioning": True,
         "version-tolerant": True,
+        "generation-subdir": None,  # subdirectory to generate the code in
     }
 
     def __init__(self, options: Optional[Dict[str, Any]] = None) -> None:

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

@@ -3,6 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
+from pathlib import Path
 from typing import List, Dict, Any, Set, Union, Literal, Optional, cast
 
 from .base import BaseType
@@ -452,3 +453,34 @@ def company_name(self) -> str:
             return self.yaml_data.get("licenseInfo", {}).get("company", "")
         # typespec azure case without custom license and swagger case
         return "Microsoft Corporation"
+
+    def get_root_dir(self) -> Path:
+        if self.options["no-namespace-folders"]:
+            # when output folder contains parts different from the namespace, we fall back to current folder directly.
+            return Path(".")
+        return Path(*self.namespace.split("."))
+
+    def get_generation_dir(self, namespace: str) -> Path:
+        """The directory to generate the code in.
+
+        If 'generation-subdir' is specified, it will be used as a subdirectory.
+        """
+        root_dir = self.get_root_dir()
+        retval = self._get_relative_generation_dir(root_dir, namespace)
+        return retval
+
+    def _get_relative_generation_dir(self, root_dir: Path, namespace: str) -> Path:
+        if self.options["no-namespace-folders"]:
+            return Path(".")
+        if self.options.get("generation-subdir"):
+            # For the main namespace, return root_dir + generation-subdir
+            if namespace in ("", self.namespace):
+                return root_dir / self.options["generation-subdir"]
+
+            # For subnamespaces, extract the subnamespace part and append it to generation-subdir
+            if namespace.startswith(self.namespace + "."):
+                subnamespace_parts = namespace[len(self.namespace) + 1 :].split(".")
+                return root_dir / self.options["generation-subdir"] / Path(*subnamespace_parts)
+
+        # No generation-subdir specified, use the namespace path directly
+        return Path(*namespace.split("."))

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

@@ -110,7 +110,9 @@ def keep_version_file(self) -> bool:
             return True
         # If the version file is already there and the version is greater than the current version, keep it.
         try:
-            serialized_version_file = self.read_file(self.exec_path(self.code_model.namespace) / "_version.py")
+            serialized_version_file = self.read_file(
+                self.code_model.get_generation_dir(self.code_model.namespace) / "_version.py"
+            )
             match = re.search(r'VERSION\s*=\s*"([^"]+)"', str(serialized_version_file))
             serialized_version = match.group(1) if match else ""
         except (FileNotFoundError, IndexError):
@@ -133,37 +135,37 @@ def serialize(self) -> None:
 
         general_serializer = GeneralSerializer(code_model=self.code_model, env=env, async_mode=False)
         for client_namespace, client_namespace_type in self.code_model.client_namespace_types.items():
-            exec_path = self.exec_path(client_namespace)
+            generation_path = self.code_model.get_generation_dir(client_namespace)
             if client_namespace == "":
                 if self.code_model.options["basic-setup-py"]:
                     # Write the setup file
-                    self.write_file(exec_path / Path("setup.py"), general_serializer.serialize_setup_file())
+                    self.write_file(generation_path / Path("setup.py"), general_serializer.serialize_setup_file())
                 elif not self.code_model.options["keep-setup-py"]:
                     # remove setup.py file
-                    self.remove_file(exec_path / Path("setup.py"))
+                    self.remove_file(generation_path / Path("setup.py"))
 
                 # add packaging files in root namespace (e.g. setup.py, README.md, etc.)
                 if self.code_model.options.get("package-mode"):
-                    self._serialize_and_write_package_files(client_namespace)
+                    self._serialize_and_write_package_files()
 
                 # write apiview-properties.json
                 if self.code_model.options.get("emit-cross-language-definition-file"):
                     self.write_file(
-                        exec_path / Path("apiview-properties.json"),
+                        generation_path / Path("apiview-properties.json"),
                         general_serializer.serialize_cross_language_definition_file(),
                     )
 
                 # add generated samples and generated tests
                 if self.code_model.options["show-operations"] and self.code_model.has_operations:
                     if self.code_model.options["generate-sample"]:
-                        self._serialize_and_write_sample(env, namespace=client_namespace)
+                        self._serialize_and_write_sample(env)
                     if self.code_model.options["generate-test"]:
-                        self._serialize_and_write_test(env, namespace=client_namespace)
+                        self._serialize_and_write_test(env)
 
                 # add _metadata.json
                 if self.code_model.metadata:
                     self.write_file(
-                        exec_path / Path("_metadata.json"),
+                        generation_path / Path("_metadata.json"),
                         json.dumps(self.code_model.metadata, indent=2),
                     )
             elif client_namespace_type.clients:
@@ -172,7 +174,7 @@ def serialize(self) -> None:
             else:
                 # add pkgutil init file if no clients in this namespace
                 self.write_file(
-                    exec_path / Path("__init__.py"),
+                    generation_path / Path("__init__.py"),
                     general_serializer.serialize_pkgutil_init_file(),
                 )
 
@@ -194,7 +196,7 @@ def serialize(self) -> None:
 
             if not self.code_model.options["models-mode"]:
                 # keep models file if users ended up just writing a models file
-                model_path = exec_path / Path("models.py")
+                model_path = generation_path / Path("models.py")
                 if self.read_file(model_path):
                     self.write_file(model_path, self.read_file(model_path))
 
@@ -210,12 +212,15 @@ def serialize(self) -> None:
             # to make sure all generated files could be packed into .zip/.whl/.tgz package
             if not client_namespace_type.clients and client_namespace_type.operation_groups and self.has_aio_folder:
                 self.write_file(
-                    exec_path / Path("aio/__init__.py"),
+                    generation_path / Path("aio/__init__.py"),
                     general_serializer.serialize_pkgutil_init_file(),
                 )
 
-    def _serialize_and_write_package_files(self, client_namespace: str) -> None:
-        root_of_sdk = self.exec_path(client_namespace)
+    def _serialize_and_write_package_files(self) -> None:
+        root_of_sdk = Path(".")
+        if self.code_model.options["no-namespace-folders"]:
+            compensation = Path("../" * (self.code_model.namespace.count(".") + 1))
+            root_of_sdk = root_of_sdk / compensation
         if self.code_model.options["package-mode"] in VALID_PACKAGE_MODE:
             env = Environment(
                 loader=PackageLoader("pygen.codegen", "templates/packaging_templates"),
@@ -266,7 +271,7 @@ def _serialize_and_write_models_folder(
         self, env: Environment, namespace: str, models: List[ModelType], enums: List[EnumType]
     ) -> None:
         # Write the models folder
-        models_path = self.exec_path(namespace) / "models"
+        models_path = self.code_model.get_generation_dir(namespace) / "models"
         serializer = DpgModelSerializer if self.code_model.options["models-mode"] == "dpg" else MsrestModelSerializer
         if self.code_model.has_non_json_models(models):
             self.write_file(
@@ -334,15 +339,15 @@ def _serialize_and_write_operations_folder(
         self, operation_groups: List[OperationGroup], env: Environment, namespace: str
     ) -> None:
         operations_folder_name = self.code_model.operations_folder_name(namespace)
-        exec_path = self.exec_path(namespace)
+        generation_path = self.code_model.get_generation_dir(namespace)
         for async_mode, async_path in self.serialize_loop:
             prefix_path = f"{async_path}{operations_folder_name}"
             # write init file
             operations_init_serializer = OperationsInitSerializer(
                 code_model=self.code_model, operation_groups=operation_groups, env=env, async_mode=async_mode
             )
             self.write_file(
-                exec_path / Path(f"{prefix_path}/__init__.py"),
+                generation_path / Path(f"{prefix_path}/__init__.py"),
                 operations_init_serializer.serialize(),
             )
 
@@ -361,26 +366,25 @@ def _serialize_and_write_operations_folder(
                     client_namespace=namespace,
                 )
                 self.write_file(
-                    exec_path / Path(f"{prefix_path}/{filename}.py"),
+                    generation_path / Path(f"{prefix_path}/{filename}.py"),
                     operation_group_serializer.serialize(),
                 )
 
             # if there was a patch file before, we keep it
-            self._keep_patch_file(exec_path / Path(f"{prefix_path}/_patch.py"), env)
+            self._keep_patch_file(generation_path / Path(f"{prefix_path}/_patch.py"), env)
 
     def _serialize_and_write_version_file(
         self,
-        namespace: str,
         general_serializer: GeneralSerializer,
     ):
-        exec_path = self.exec_path(namespace)
+        generation_path = self.code_model.get_root_dir()
 
         def _read_version_file(original_version_file_name: str) -> str:
-            return self.read_file(exec_path / original_version_file_name)
+            return self.read_file(generation_path / original_version_file_name)
 
         def _write_version_file(original_version_file_name: str) -> None:
             self.write_file(
-                exec_path / Path("_version.py"),
+                generation_path / Path("_version.py"),
                 _read_version_file(original_version_file_name),
             )
 
@@ -390,7 +394,7 @@ def _write_version_file(original_version_file_name: str) -> None:
             _write_version_file(original_version_file_name="version.py")
         elif self.code_model.options.get("package-version"):
             self.write_file(
-                exec_path / Path("_version.py"),
+                generation_path / Path("_version.py"),
                 general_serializer.serialize_version_file(),
             )
 
@@ -400,47 +404,47 @@ def _serialize_client_and_config_files(
         clients: List[Client],
         env: Environment,
     ) -> None:
-        exec_path = self.exec_path(namespace)
+        generation_path = self.code_model.get_generation_dir(namespace)
         for async_mode, async_path in self.serialize_loop:
             general_serializer = GeneralSerializer(
                 code_model=self.code_model, env=env, async_mode=async_mode, client_namespace=namespace
             )
             # when there is client.py, there must be __init__.py
             self.write_file(
-                exec_path / Path(f"{async_path}__init__.py"),
+                generation_path / Path(f"{async_path}__init__.py"),
                 general_serializer.serialize_init_file([c for c in clients if c.has_operations]),
             )
 
             # if there was a patch file before, we keep it
-            self._keep_patch_file(exec_path / Path(f"{async_path}_patch.py"), env)
+            self._keep_patch_file(generation_path / Path(f"{async_path}_patch.py"), env)
 
             if self.code_model.clients_has_operations(clients):
 
                 # write client file
                 self.write_file(
-                    exec_path / Path(f"{async_path}{self.code_model.client_filename}.py"),
+                    generation_path / Path(f"{async_path}{self.code_model.client_filename}.py"),
                     general_serializer.serialize_service_client_file(clients),
                 )
 
                 # write config file
                 self.write_file(
-                    exec_path / Path(f"{async_path}_configuration.py"),
+                    generation_path / Path(f"{async_path}_configuration.py"),
                     general_serializer.serialize_config_file(clients),
                 )
 
                 # sometimes we need define additional Mixin class for client in _utils.py
                 self._serialize_and_write_utils_folder(env, namespace)
 
-    def _serialize_and_write_utils_folder(self, env: Environment, namespace: str) -> None:
-        exec_path = self.exec_path(namespace)
+    def _serialize_and_write_utils_folder(self, env: Environment, namespace: str):
+        generation_dir = self.code_model.get_generation_dir(namespace)
         general_serializer = GeneralSerializer(code_model=self.code_model, env=env, async_mode=False)
-        utils_folder_path = exec_path / Path("_utils")
-        if self.code_model.need_utils_folder(async_mode=False, client_namespace=namespace):
+        utils_folder_path = generation_dir / Path("_utils")
+        if self.code_model.need_utils_folder(async_mode=False, client_namespace=self.code_model.namespace):
             self.write_file(
                 utils_folder_path / Path("__init__.py"),
                 self.code_model.license_header,
             )
-        if self.code_model.need_utils_utils(async_mode=False, client_namespace=namespace):
+        if self.code_model.need_utils_utils(async_mode=False, client_namespace=self.code_model.namespace):
             self.write_file(
                 utils_folder_path / Path("utils.py"),
                 general_serializer.need_utils_utils_file(),
@@ -460,60 +464,43 @@ def _serialize_and_write_utils_folder(self, env: Environment, namespace: str) ->
             )
 
     def _serialize_and_write_top_level_folder(self, env: Environment, namespace: str) -> None:
-        exec_path = self.exec_path(namespace)
+        root_dir = self.code_model.get_root_dir()
         # write _utils folder
-        self._serialize_and_write_utils_folder(env, namespace)
+        self._serialize_and_write_utils_folder(env, self.code_model.namespace)
 
         general_serializer = GeneralSerializer(code_model=self.code_model, env=env, async_mode=False)
 
         # write _version.py
-        self._serialize_and_write_version_file(namespace, general_serializer)
+        self._serialize_and_write_version_file(general_serializer)
 
         # write the empty py.typed file
-        self.write_file(exec_path / Path("py.typed"), "# Marker file for PEP 561.")
+        pytyped_value = "# Marker file for PEP 561."
+        # TODO: remove this when we remove legacy multiapi generation
+        if self.code_model.options["multiapi"]:
+            self.write_file(self.code_model.get_generation_dir(namespace) / Path("py.typed"), pytyped_value)
+        else:
+            self.write_file(root_dir / Path("py.typed"), pytyped_value)
 
         # write _validation.py
         if any(og for client in self.code_model.clients for og in client.operation_groups if og.need_validation):
             self.write_file(
-                exec_path / Path("_validation.py"),
+                root_dir / Path("_validation.py"),
                 general_serializer.serialize_validation_file(),
             )
 
         # write _types.py
         if self.code_model.named_unions:
             self.write_file(
-                exec_path / Path("_types.py"),
+                root_dir / Path("_types.py"),
                 TypesSerializer(code_model=self.code_model, env=env).serialize(),
             )
 
     def _serialize_and_write_metadata(self, env: Environment, namespace: str) -> None:
-        metadata_serializer = MetadataSerializer(self.code_model, env, client_namespace=namespace)
-        self.write_file(self.exec_path(namespace) / Path("_metadata.json"), metadata_serializer.serialize())
-
-    @property
-    def exec_path_compensation(self) -> Path:
-        """Assume the process is running in the root folder of the package. If not, we need the path compensation."""
-        return (
-            Path("../" * (self.code_model.namespace.count(".") + 1))
-            if self.code_model.options["no-namespace-folders"]
-            else Path(".")
+        metadata_serializer = MetadataSerializer(self.code_model, env)
+        self.write_file(
+            self.code_model.get_generation_dir(namespace) / Path("_metadata.json"), metadata_serializer.serialize()
         )
 
-    def exec_path_for_test_sample(self, namespace: str) -> Path:
-        return self.exec_path_compensation / Path(*namespace.split("."))
-
-    # pylint: disable=line-too-long
-    def exec_path(self, namespace: str) -> Path:
-        if (
-            self.code_model.options["no-namespace-folders"]
-            and not self.code_model.options["multiapi"]
-            and not self.code_model.options["azure-arm"]
-        ):
-            # when output folder contains parts different from the namespace, we fall back to current folder directly.
-            # (e.g. https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/communication/azure-communication-callautomation/swagger/SWAGGER.md)
-            return Path(".")
-        return self.exec_path_compensation / Path(*namespace.split("."))
-
     # pylint: disable=line-too-long
     @property
     def sample_additional_folder(self) -> Path:
@@ -532,8 +519,8 @@ def sample_additional_folder(self) -> Path:
             return Path("/".join(namespace_config.split(".")[num_of_package_namespace:]))
         return Path("")
 
-    def _serialize_and_write_sample(self, env: Environment, namespace: str):
-        out_path = self.exec_path_for_test_sample(namespace) / Path("generated_samples")
+    def _serialize_and_write_sample(self, env: Environment):
+        out_path = Path("./generated_samples")
         for client in self.code_model.clients:
             for op_group in client.operation_groups:
                 for operation in op_group.operations:
@@ -565,9 +552,9 @@ def _serialize_and_write_sample(self, env: Environment, namespace: str):
                             log_error = f"error happens in sample {file}: {e}"
                             _LOGGER.error(log_error)
 
-    def _serialize_and_write_test(self, env: Environment, namespace: str):
+    def _serialize_and_write_test(self, env: Environment):
         self.code_model.for_test = True
-        out_path = self.exec_path_for_test_sample(namespace) / Path("generated_tests")
+        out_path = Path("./generated_tests")
         general_serializer = TestGeneralSerializer(code_model=self.code_model, env=env)
         self.write_file(out_path / "conftest.py", general_serializer.serialize_conftest())
         if not self.code_model.options["azure-arm"]: