feat: reduced generated file size ~50%

Author: Anders Brams

.github/workflows/qa.yml

@@ -82,3 +82,21 @@ jobs:
           uv run python scripts/benchmark_generate.py compare --baseline
           .benchmark/base.json --candidate .benchmark/current.json
           --max-regression 0.02
+
+      - name: Measure generated file sizes (base)
+        run: >
+          uv run python scripts/benchmark_generate.py size --package-path
+          .benchmark/base --spec tests/performance/nautobot.json.gz
+          --output .benchmark/base-size.json
+
+      - name: Measure generated file sizes (current)
+        run: >
+          uv run python scripts/benchmark_generate.py size --package-path .
+          --spec tests/performance/nautobot.json.gz
+          --output .benchmark/current-size.json
+
+      - name: Check generated file size regression
+        run: >
+          uv run python scripts/benchmark_generate.py compare-size --baseline
+          .benchmark/base-size.json --candidate .benchmark/current-size.json
+          --max-regression 0.02

openapi_python/generator/render.py

@@ -54,6 +54,22 @@ def _field_annotation(field: FieldDef) -> str:
     return annotation
 
 
+def _class_field_annotation(field: FieldDef, total_optional: bool) -> str:
+    annotation = _render_annotation(field.annotation)
+    if not field.required and not total_optional:
+        annotation = f"NotRequired[{annotation}]"
+    return annotation
+
+
+def _supports_typeddict_class_syntax(defn: TypedDictDef) -> bool:
+    return all(
+        field.name.isidentifier()
+        and not keyword.iskeyword(field.name)
+        and not field.name.startswith("__")
+        for field in defn.fields
+    )
+
+
 _TEMPLATE_DIR = Path(__file__).with_name("templates")
 _JINJA_ENV = Environment(
     loader=FileSystemLoader(_TEMPLATE_DIR),
@@ -64,6 +80,7 @@ def _field_annotation(field: FieldDef) -> str:
 _JINJA_ENV.filters["repr"] = repr
 _JINJA_ENV.filters["annotation"] = _render_annotation
 _JINJA_ENV.filters["field_annotation"] = _field_annotation
+_JINJA_ENV.filters["class_field_annotation"] = _class_field_annotation
 
 
 def _render_template(name: str, **context: object) -> str:
@@ -76,7 +93,15 @@ def _indent(text: str, spaces: int = 4) -> str:
 
 
 def _format_typeddict(defn: TypedDictDef) -> str:
-    return _render_template("typeddict.py.j2", defn=defn)
+    total_optional = bool(defn.fields) and all(
+        not field.required for field in defn.fields
+    )
+    return _render_template(
+        "typeddict.py.j2",
+        defn=defn,
+        class_syntax=_supports_typeddict_class_syntax(defn),
+        total_optional=total_optional,
+    )
 
 
 def _format_alias(alias: TypeAliasDef) -> str:
@@ -249,22 +274,13 @@ def _method_overload_line(op: OperationDef, *, is_async: bool = False) -> str:
     )
 
 
-def _method_dispatch_line(op: OperationDef, *, is_async: bool = False) -> str:
-    return _render_template(
-        "method_dispatch.py.j2",
-        op=op,
-        is_async=is_async,
-    )
-
-
 def _fallback_method_block(
-    method: str, overloads: list[str], dispatch: list[str], *, is_async: bool = False
+    method: str, overloads: list[str], *, is_async: bool = False
 ) -> str:
     return _render_template(
         "method_block.py.j2",
         method=method,
         overloads="\n".join(overloads),
-        dispatch_block="\n\n        ".join(dispatch),
         callable_return="Awaitable[Any]" if is_async else "object",
         call_return="Any" if is_async else "object",
         is_async=is_async,
@@ -330,27 +346,20 @@ def _render_client(spec: NormalizedSpec, *, transport_mode: str) -> str:
     async_protocols: list[str] = []
     method_overloads: dict[str, list[str]] = {}
     async_method_overloads: dict[str, list[str]] = {}
-    method_dispatch: dict[str, list[str]] = {}
-    async_method_dispatch: dict[str, list[str]] = {}
     for op in spec.operations:
         protocols.append(_protocol_block(op))
         async_protocols.append(_protocol_block(op, is_async=True))
         method_overloads.setdefault(op.method, []).append(_method_overload_line(op))
         async_method_overloads.setdefault(op.method, []).append(
             _method_overload_line(op, is_async=True)
         )
-        method_dispatch.setdefault(op.method, []).append(_method_dispatch_line(op))
-        async_method_dispatch.setdefault(op.method, []).append(
-            _method_dispatch_line(op, is_async=True)
-        )
 
     method_blocks: list[str] = []
     for method in sorted(method_overloads):
         method_blocks.append(
             _fallback_method_block(
                 method,
                 method_overloads[method],
-                method_dispatch.get(method, []),
             )
         )
     async_method_blocks: list[str] = []
@@ -359,7 +368,6 @@ def _render_client(spec: NormalizedSpec, *, transport_mode: str) -> str:
             _fallback_method_block(
                 method,
                 async_method_overloads[method],
-                async_method_dispatch.get(method, []),
                 is_async=True,
             )
         )

openapi_python/generator/templates/method_block.py.j2

@@ -2,15 +2,14 @@
 @overload
 def {{ method }}(self, route: str) -> Callable[..., {{ callable_return }}]: ...
 def {{ method }}(self, route: str) -> Callable[..., {{ callable_return }}]:
-        {{ dispatch_block }}
-        {{ "async " if is_async else "" }}def _call(*, params: dict[str, object] | None = None, query: dict[str, object] | None = None, headers: dict[str, object] | None = None, body: object | None = None) -> {{ call_return }}:
-            return {{ "await " if is_async else "" }}self._transport.request(
-                method={{ method|repr }},
-                route=route,
-                base_url=self._base_url,
-                params=params,
-                query=query,
-                headers=headers,
-                body=body,
-            )
-        return _call
+    {{ "async " if is_async else "" }}def _call(*, params: dict[str, object] | None = None, query: dict[str, object] | None = None, headers: dict[str, object] | None = None, body: object | None = None) -> {{ call_return }}:
+        return {{ "await " if is_async else "" }}self._transport.request(
+            method={{ method|repr }},
+            route=route,
+            base_url=self._base_url,
+            params=params,
+            query=query,
+            headers=headers,
+            body=body,
+        )
+    return _call

openapi_python/generator/templates/typeddict.py.j2

@@ -1,4 +1,13 @@
-{% if not defn.fields -%}
+{% if class_syntax -%}
+class {{ defn.name }}(TypedDict{% if total_optional %}, total=False{% endif %}):
+{% if not defn.fields %}
+    pass
+{% else %}
+{% for field in defn.fields %}
+    {{ field.name }}: {{ field | class_field_annotation(total_optional) }}
+{% endfor %}
+{% endif %}
+{% elif not defn.fields -%}
 {{ defn.name }} = TypedDict({{ defn.name | repr }}, {})
 {% else -%}
 {{ defn.name }} = TypedDict(

pyproject.toml

@@ -42,7 +42,7 @@ dev = [
     "python-multipart>=0.0.20",
     "ruff>=0.9.10",
     "twine>=6.1.0",
-    "ty>=0.0.34",
+    "ty>=0.0.37",
     "uvicorn>=0.34.0",
 ]
 

scripts/benchmark_generate.py

@@ -50,6 +50,32 @@ def _run_once(
         shutil.rmtree(output_dir, ignore_errors=True)
 
 
+def _run_size_once(
+    *,
+    generate_client: Any,
+    generation_request: Any,
+    spec_json: str,
+    package_name: str,
+) -> tuple[dict[str, int], Any]:
+    output_dir = Path(tempfile.mkdtemp(prefix="openapi-python-size-"))
+    try:
+        result = generate_client(
+            generation_request(
+                output_dir=output_dir,
+                spec_json=spec_json,
+                package_name=package_name,
+                overwrite=True,
+            )
+        )
+        file_sizes = {
+            path.relative_to(output_dir).as_posix(): path.stat().st_size
+            for path in result.written_files
+        }
+        return file_sizes, result
+    finally:
+        shutil.rmtree(output_dir, ignore_errors=True)
+
+
 def run_benchmark(args: argparse.Namespace) -> int:
     spec_json = _load_spec(args.spec)
     generation_request, generate_client = _load_generator(args.package_path)
@@ -93,6 +119,35 @@ def run_benchmark(args: argparse.Namespace) -> int:
     return 0
 
 
+def run_size_benchmark(args: argparse.Namespace) -> int:
+    spec_json = _load_spec(args.spec)
+    generation_request, generate_client = _load_generator(args.package_path)
+
+    file_sizes, result = _run_size_once(
+        generate_client=generate_client,
+        generation_request=generation_request,
+        spec_json=spec_json,
+        package_name=args.package,
+    )
+
+    payload = {
+        "files_bytes": file_sizes,
+        "operations": result.operations,
+        "total_bytes": sum(file_sizes.values()),
+        "type_definitions": result.type_definitions,
+    }
+
+    encoded = json.dumps(payload, indent=2, sort_keys=True)
+    if args.output:
+        args.output.write_text(encoded + "\n", encoding="utf-8")
+    print(encoded)
+    return 0
+
+
+def _format_bytes(value: float) -> str:
+    return f"{value:,.0f} bytes"
+
+
 def compare_benchmarks(args: argparse.Namespace) -> int:
     baseline = json.loads(args.baseline.read_text(encoding="utf-8"))
     candidate = json.loads(args.candidate.read_text(encoding="utf-8"))
@@ -116,6 +171,29 @@ def compare_benchmarks(args: argparse.Namespace) -> int:
     return 0
 
 
+def compare_size_benchmarks(args: argparse.Namespace) -> int:
+    baseline = json.loads(args.baseline.read_text(encoding="utf-8"))
+    candidate = json.loads(args.candidate.read_text(encoding="utf-8"))
+
+    baseline_bytes = float(baseline["total_bytes"])
+    candidate_bytes = float(candidate["total_bytes"])
+    allowed_bytes = baseline_bytes * (1 + args.max_regression)
+    change = (candidate_bytes - baseline_bytes) / baseline_bytes
+
+    print(f"baseline total:  {_format_bytes(baseline_bytes)}")
+    print(f"candidate total: {_format_bytes(candidate_bytes)}")
+    print(f"change:          {change:+.2%}")
+    print(f"limit:           +{args.max_regression:.2%}")
+
+    if candidate_bytes > allowed_bytes:
+        print(
+            "generated file size regressed beyond the configured limit",
+            file=sys.stderr,
+        )
+        return 1
+    return 0
+
+
 def _build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         prog="benchmark_generate.py",
@@ -132,12 +210,27 @@ def _build_parser() -> argparse.ArgumentParser:
     run.add_argument("--output", type=Path)
     run.set_defaults(func=run_benchmark)
 
+    size = subcommands.add_parser("size", help="Measure generated file sizes")
+    size.add_argument("--spec", type=Path, required=True)
+    size.add_argument("--package-path", type=Path, default=Path.cwd())
+    size.add_argument("--package", default="my_client")
+    size.add_argument("--output", type=Path)
+    size.set_defaults(func=run_size_benchmark)
+
     compare = subcommands.add_parser("compare", help="Compare two benchmark results")
     compare.add_argument("--baseline", type=Path, required=True)
     compare.add_argument("--candidate", type=Path, required=True)
     compare.add_argument("--max-regression", type=float, default=0.02)
     compare.set_defaults(func=compare_benchmarks)
 
+    compare_size = subcommands.add_parser(
+        "compare-size", help="Compare two generated file size results"
+    )
+    compare_size.add_argument("--baseline", type=Path, required=True)
+    compare_size.add_argument("--candidate", type=Path, required=True)
+    compare_size.add_argument("--max-regression", type=float, default=0.02)
+    compare_size.set_defaults(func=compare_size_benchmarks)
+
     return parser