Merge pull request #11 from mattcoulter7/feat/openapi-overlay

Feat/openapi overlay

Author: mattcoulter7

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pydantic-openapi-generator"
-version = "2.5.1"
+version = "2.6.0"
 description = "Openapi Python Generator"
 authors = [
     { name = "Marco Müllner", email = "muellnermarco@gmail.com" },

src/pydantic_openapi_generator/__main__.py

@@ -58,6 +58,13 @@
     default=None,
     help="Optional YAML configuration for generated client parameters.",
 )
+@click.option(
+    "--overlay",
+    "overlay_paths",
+    type=click.Path(exists=True, dir_okay=False),
+    multiple=True,
+    help="Optional JSON/YAML OpenAPI overlay file. Can be provided multiple times and is applied in order.",
+)
 @click.version_option(version=__version__)
 def main(
     source: str,
@@ -69,6 +76,7 @@ def main(
     pydantic_version: PydanticVersion = PydanticVersion.V2,
     formatter: Formatter = Formatter.BLACK,
     config_path: Optional[str] = None,
+    overlay_paths: tuple[str, ...] = (),
 ) -> None:
     """
     Generate Python code from an OpenAPI 3.0+ specification.
@@ -86,6 +94,7 @@ def main(
         pydantic_version,
         formatter,
         config_path,
+        overlay_paths,
     )
 
 

src/pydantic_openapi_generator/generate_data.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import List, Optional, Union
+from typing import Any, List, Optional, Sequence, Union
 
 import black
 import click
@@ -53,14 +53,54 @@ def format_using_black(content: str) -> str:
     return isort.code(formatted_contend, line_length=FormatOptions.line_length)
 
 
-def get_open_api(source: Union[str, Path]):
+def load_openapi_data(source: Union[str, Path]) -> dict[str, Any]:
+    """
+    Load an OpenAPI JSON/YAML document from a URL or local file path.
+    """
+    if not isinstance(source, Path) and (source.startswith("http://") or source.startswith("https://")):
+        content = httpx.get(source).text
+    else:
+        with open(source, "r") as f:
+            content = f.read()
+
+    try:
+        data = orjson.loads(content)
+    except orjson.JSONDecodeError:
+        try:
+            data = yaml.safe_load(content)
+        except yaml.YAMLError as e:
+            click.echo(f"File {source} is neither a valid JSON nor YAML file: {str(e)}")
+            raise
+
+    if not isinstance(data, dict):
+        raise ValueError(f"OpenAPI data loaded from {source} must be a JSON/YAML object.")
+
+    return data
+
+
+def deep_merge(base: dict[str, Any], overlay: dict[str, Any]) -> dict[str, Any]:
+    """
+    Recursively merge two dictionaries. Overlay lists and scalar values replace base values.
+    """
+    result = base.copy()
+    for key, overlay_value in overlay.items():
+        base_value = result.get(key)
+        if isinstance(base_value, dict) and isinstance(overlay_value, dict):
+            result[key] = deep_merge(base_value, overlay_value)
+        else:
+            result[key] = overlay_value
+    return result
+
+
+def get_open_api(source: Union[str, Path], overlay_paths: Optional[Sequence[Union[str, Path]]] = None):
     """
     Tries to fetch the openapi specification file from the web or load from a local file.
     Supports both JSON and YAML formats. Returns the according OpenAPI object.
     Automatically supports OpenAPI 3.0 and 3.1 specifications with intelligent version detection.
 
     Args:
         source: URL or file path to the OpenAPI specification
+        overlay_paths: Optional JSON/YAML overlay files to deep-merge into source in order
 
     Returns:
         tuple: (OpenAPI object, version) where version is "3.0" or "3.1"
@@ -72,29 +112,9 @@ def get_open_api(source: Union[str, Path]):
         JSONDecodeError/YAMLError: If the file cannot be parsed
     """
     try:
-        # Handle remote files
-        if not isinstance(source, Path) and (source.startswith("http://") or source.startswith("https://")):
-            content = httpx.get(source).text
-            # Try JSON first, then YAML for remote files
-            try:
-                data = orjson.loads(content)
-            except orjson.JSONDecodeError:
-                data = yaml.safe_load(content)
-        else:
-            # Handle local files
-            with open(source, "r") as f:
-                file_content = f.read()
-
-                # Try JSON first
-                try:
-                    data = orjson.loads(file_content)
-                except orjson.JSONDecodeError:
-                    # If JSON fails, try YAML
-                    try:
-                        data = yaml.safe_load(file_content)
-                    except yaml.YAMLError as e:
-                        click.echo(f"File {source} is neither a valid JSON nor YAML file: {str(e)}")
-                        raise
+        data = load_openapi_data(source)
+        for overlay_path in overlay_paths or ():
+            data = deep_merge(data, load_openapi_data(overlay_path))
 
         # Detect version and parse with appropriate parser
         version = detect_openapi_version(data)
@@ -206,12 +226,13 @@ def generate_data(
     pydantic_version: PydanticVersion = PydanticVersion.V2,
     formatter: Formatter = Formatter.BLACK,
     config_path: Optional[Union[str, Path]] = None,
+    overlay_paths: Optional[Sequence[Union[str, Path]]] = None,
     config: Optional[PydanticOpenAPIGeneratorConfig] = None,
 ) -> None:
     """
     Generate Python code from an OpenAPI 3.0+ specification.
     """
-    openapi_obj, version = get_open_api(source)
+    openapi_obj, version = get_open_api(source, overlay_paths)
     loaded_config = config if config is not None else load_config(config_path)
     click.echo(f"Generating data from {source} (OpenAPI {version})")
 

tests/test_generate_data.py

@@ -1,6 +1,7 @@
-from pathlib import Path
+import json
 import shutil
 import subprocess
+from pathlib import Path
 
 import orjson
 import pytest
@@ -53,6 +54,62 @@ def test_get_open_api(model_data):
         get_open_api(test_data_folder / "file_does_not_exist.json")
 
 
+def test_get_open_api_deep_merges_json_and_yaml_overlays_in_order(tmp_path):
+    source_path = tmp_path / "openapi.yaml"
+    first_overlay_path = tmp_path / "first-overlay.json"
+    second_overlay_path = tmp_path / "second-overlay.yaml"
+    source_path.write_text(
+        """
+openapi: 3.0.1
+info:
+    title: Overlay test
+    version: "1"
+paths: {}
+components:
+    schemas:
+        CommunicationDetails:
+            type: object
+            required:
+                - received_date
+                - sent_date
+            properties:
+                received_date:
+                    type: string
+                    format: date-time
+                sent_date:
+                    type: string
+                    format: date-time
+"""
+    )
+    first_overlay_path.write_text(
+        json.dumps(
+            {
+                "components": {
+                    "schemas": {
+                        "CommunicationDetails": {"required": ["received_date"]},
+                    }
+                }
+            }
+        )
+    )
+    second_overlay_path.write_text(
+        """
+components:
+    schemas:
+        CommunicationDetails:
+            required:
+                - sent_date
+"""
+    )
+
+    openapi_obj, version = get_open_api(source_path, [first_overlay_path, second_overlay_path])
+
+    communication_details = openapi_obj.components.schemas["CommunicationDetails"]
+    assert version == "3.0"
+    assert communication_details.required == ["sent_date"]
+    assert set(communication_details.properties) == {"received_date", "sent_date"}
+
+
 def test_generate_data(model_data_with_cleanup):
     generate_data(test_data_path, test_result_path)
     assert test_result_path.exists()

tests/test_main.py

@@ -1,6 +1,7 @@
 """Test cases for the __main__ module."""
 
 import json
+from unittest.mock import patch
 
 import pytest
 from click.testing import CliRunner
@@ -55,3 +56,29 @@ def test_main_accepts_config_path(runner: CliRunner, tmp_path) -> None:
 
     assert result.exit_code == 0
     assert (output_path / "clients" / "sync_client.py").exists()
+
+
+def test_main_accepts_multiple_overlays_in_order(runner: CliRunner, tmp_path) -> None:
+    source_path = tmp_path / "openapi.json"
+    output_path = tmp_path / "generated"
+    first_overlay_path = tmp_path / "first.yaml"
+    second_overlay_path = tmp_path / "second.json"
+    source_path.write_text(json.dumps(CONTRACT_SPEC))
+    first_overlay_path.write_text("components: {}")
+    second_overlay_path.write_text("{}")
+
+    with patch("pydantic_openapi_generator.__main__.generate_data") as generate_data_mock:
+        result = runner.invoke(
+            main,
+            [
+                str(source_path),
+                str(output_path),
+                "--overlay",
+                str(first_overlay_path),
+                "--overlay",
+                str(second_overlay_path),
+            ],
+        )
+
+    assert result.exit_code == 0
+    assert generate_data_mock.call_args.args[-1] == (str(first_overlay_path), str(second_overlay_path))