Add file:// URL protocol support for $ref resolution (#2601)

* Add support for file URL scheme in schema fetching and validation

* Add support for UNC paths in file URL handling

* Add support for relative path resolution in file URL handling

* Add fragment handling and improve file URL joining logic

* Add tests for handling file URLs with UNC and local paths

* Refactor tests to assert path components for file URL handling

* Refactor tests to assert path components for file URL handling

Author: koxudaxi

src/datamodel_code_generator/__main__.py

@@ -166,7 +166,7 @@ def validate_url(cls, value: Any) -> ParseResult | None:  # noqa: N805
             return urlparse(value)
         if value is None:  # pragma: no cover
             return None
-        msg = f"This protocol doesn't support only http/https. --input={value}"  # pragma: no cover
+        msg = f"Unsupported URL scheme. Supported: http, https, file. --input={value}"  # pragma: no cover
         raise Error(msg)  # pragma: no cover
 
     # Pydantic 1.5.1 doesn't support each_item=True correctly

src/datamodel_code_generator/http.py

@@ -1,21 +1,26 @@
 """HTTP utilities for fetching remote schema files.
 
 Provides functions to fetch schema content from URLs and join URL references.
-Requires the 'http' extra: `pip install 'datamodel-code-generator[http]'`.
+HTTP(S) URLs require the 'http' extra: `pip install 'datamodel-code-generator[http]'`.
+file:// URLs are handled without additional dependencies.
 """
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
 
-try:
-    import httpx
-except ImportError as exc:  # pragma: no cover
-    msg = "Please run `$pip install 'datamodel-code-generator[http]`' to resolve URL Reference"
-    raise Exception(msg) from exc  # noqa: TRY002
+
+def _get_httpx() -> Any:
+    """Lazily import httpx, raising a helpful error if not installed."""
+    try:
+        import httpx  # noqa: PLC0415
+    except ImportError as exc:  # pragma: no cover
+        msg = "Please run `$pip install 'datamodel-code-generator[http]`' to resolve HTTP(S) URL references"
+        raise Exception(msg) from exc  # noqa: TRY002
+    return httpx
 
 
 def get_body(
@@ -25,6 +30,7 @@ def get_body(
     query_parameters: Sequence[tuple[str, str]] | None = None,
 ) -> str:
     """Fetch content from a URL with optional headers and query parameters."""
+    httpx = _get_httpx()
     return httpx.get(
         url,
         headers=headers,
@@ -35,6 +41,51 @@ def get_body(
     ).text
 
 
-def join_url(url: str, ref: str = ".") -> str:
+def join_url(url: str, ref: str = ".") -> str:  # noqa: PLR0912
     """Join a base URL with a relative reference."""
+    if url.startswith("file://"):
+        from urllib.parse import urlparse  # noqa: PLC0415
+
+        parsed = urlparse(url)
+
+        if ref.startswith("file://"):
+            return ref
+
+        ref_path, *frag = ref.split("#", 1)
+
+        # Fragment-only ref: keep the original path
+        if not ref_path:
+            joined = url.split("#", maxsplit=1)[0]
+            if frag:
+                joined += f"#{frag[0]}"
+            return joined
+
+        if ref_path.startswith("/"):
+            joined_path = ref_path
+        else:
+            base_segments = parsed.path.lstrip("/").split("/")
+            if base_segments and not base_segments[0]:
+                base_segments = []
+            if base_segments:
+                base_segments = base_segments[:-1]
+
+            min_depth = 1 if parsed.netloc else 0
+            for segment in ref_path.split("/"):
+                if segment in {"", "."}:
+                    continue
+                if segment == "..":
+                    if len(base_segments) > min_depth:
+                        base_segments.pop()
+                    continue
+                base_segments.append(segment)
+
+            joined_path = "/" + "/".join(base_segments)
+            if ref_path.endswith("/"):
+                joined_path += "/"
+
+        joined = f"file://{parsed.netloc}{joined_path}"
+        if frag:
+            joined += f"#{frag[0]}"
+        return joined
+    httpx = _get_httpx()
     return str(httpx.URL(url).join(ref))

src/datamodel_code_generator/parser/jsonschema.py

@@ -1964,8 +1964,21 @@ def _get_ref_body(self, resolved_ref: str) -> dict[str, YamlValue]:
         return self._get_ref_body_from_remote(resolved_ref)
 
     def _get_ref_body_from_url(self, ref: str) -> dict[str, YamlValue]:
-        """Get reference body from a URL."""
-        # URL Reference: $ref: 'http://path/to/your/resource' Uses the whole document located on the different server.
+        """Get reference body from a URL (HTTP, HTTPS, or file scheme)."""
+        if ref.startswith("file://"):
+            from urllib.parse import urlparse  # noqa: PLC0415
+            from urllib.request import url2pathname  # noqa: PLC0415
+
+            parsed = urlparse(ref)
+            # url2pathname handles percent-decoding and Windows drive letters
+            path = url2pathname(parsed.path)
+            # Handle UNC paths (file://server/share/path)
+            if parsed.netloc:
+                path = f"//{parsed.netloc}{path}"
+            file_path = Path(path)
+            return self.remote_object_cache.get_or_put(
+                ref, default_factory=lambda _: load_yaml_dict_from_path(file_path, self.encoding)
+            )
         return self.remote_object_cache.get_or_put(
             ref, default_factory=lambda key: load_yaml_dict(self._get_text_from_url(key))
         )

src/datamodel_code_generator/reference.py

@@ -481,7 +481,7 @@ def base_url_context(self, base_url: str | None) -> Generator[None, None, None]:
         """Temporarily set the base URL within a context.
 
         Only sets the base_url if:
-        - The new value is actually a URL (http:// or https://)
+        - The new value is actually a URL (http://, https://, or file://)
         - OR _base_url was already set (switching between URLs)
         This preserves backward compatibility for local file parsing where
         this method was previously a no-op.
@@ -858,5 +858,5 @@ def snake_to_upper_camel(word: str, delimiter: str = "_") -> str:
 
 
 def is_url(ref: str) -> bool:
-    """Check if a reference string is an HTTP(S) URL."""
-    return ref.startswith(("https://", "http://"))
+    """Check if a reference string is a URL (HTTP, HTTPS, or file scheme)."""
+    return ref.startswith(("https://", "http://", "file://"))

tests/main/jsonschema/test_main_jsonschema.py

@@ -3085,3 +3085,86 @@ def test_main_jsonschema_collapse_root_models_with_optional(output_file: Path) -
         assert_func=assert_file_content,
         extra_args=["--collapse-root-models"],
     )
+
+
+def test_main_jsonschema_file_url_ref(tmp_path: Path) -> None:
+    """Test that file:// URL $ref is resolved correctly."""
+    pet_schema = {
+        "$schema": "https://json-schema.org/draft/2020-12/schema",
+        "type": "object",
+        "properties": {"name": {"type": "string"}, "age": {"type": "integer"}},
+        "required": ["name"],
+    }
+    pet_file = tmp_path / "pet.json"
+    pet_file.write_text(json.dumps(pet_schema))
+
+    main_schema = {
+        "$schema": "https://json-schema.org/draft/2020-12/schema",
+        "type": "object",
+        "properties": {"pet": {"$ref": pet_file.as_uri()}},
+    }
+    main_file = tmp_path / "main.json"
+    main_file.write_text(json.dumps(main_schema))
+
+    expected = (
+        "# generated by datamodel-codegen:\n"
+        "#   filename:  main.json\n\n"
+        "from __future__ import annotations\n\n"
+        "from typing import Optional\n\n"
+        "from pydantic import BaseModel\n\n\n"
+        "class Pet(BaseModel):\n"
+        "    name: str\n"
+        "    age: Optional[int] = None\n\n\n"
+        "class Model(BaseModel):\n"
+        "    pet: Optional[Pet] = None\n"
+    )
+    run_main_and_assert(
+        input_path=main_file,
+        output_path=tmp_path / "output.py",
+        input_file_type="jsonschema",
+        expected_output=expected,
+        ignore_whitespace=True,
+        extra_args=["--disable-timestamp"],
+    )
+
+
+def test_main_jsonschema_file_url_ref_percent_encoded(tmp_path: Path) -> None:
+    """Test that file:// URL with percent-encoded path is resolved correctly."""
+    dir_with_space = tmp_path / "my schemas"
+    dir_with_space.mkdir()
+
+    pet_schema = {
+        "$schema": "https://json-schema.org/draft/2020-12/schema",
+        "type": "object",
+        "properties": {"name": {"type": "string"}},
+    }
+    pet_file = dir_with_space / "pet.json"
+    pet_file.write_text(json.dumps(pet_schema))
+
+    main_schema = {
+        "$schema": "https://json-schema.org/draft/2020-12/schema",
+        "type": "object",
+        "properties": {"pet": {"$ref": pet_file.as_uri()}},
+    }
+    main_file = tmp_path / "main.json"
+    main_file.write_text(json.dumps(main_schema))
+
+    expected = (
+        "# generated by datamodel-codegen:\n"
+        "#   filename:  main.json\n\n"
+        "from __future__ import annotations\n\n"
+        "from typing import Optional\n\n"
+        "from pydantic import BaseModel\n\n\n"
+        "class Pet(BaseModel):\n"
+        "    name: Optional[str] = None\n\n\n"
+        "class Model(BaseModel):\n"
+        "    pet: Optional[Pet] = None\n"
+    )
+    run_main_and_assert(
+        input_path=main_file,
+        output_path=tmp_path / "output.py",
+        input_file_type="jsonschema",
+        expected_output=expected,
+        ignore_whitespace=True,
+        extra_args=["--disable-timestamp"],
+    )

tests/parser/test_jsonschema.py

@@ -807,3 +807,40 @@ def test_create_data_model_dataclass_arguments(
     result = parser._create_data_model(**kwargs)
     assert isinstance(result, DataClass)
     assert result.dataclass_arguments == expected
+
+
+def test_get_ref_body_from_url_file_unc_path(mocker: MockerFixture) -> None:
+    """Test _get_ref_body_from_url handles UNC file:// URLs correctly."""
+    parser = JsonSchemaParser("")
+    mock_load = mocker.patch(
+        "datamodel_code_generator.parser.jsonschema.load_yaml_dict_from_path",
+        return_value={"type": "object"},
+    )
+
+    result = parser._get_ref_body_from_url("file://server/share/schemas/pet.json")
+
+    assert result == {"type": "object"}
+    mock_load.assert_called_once()
+    called_path = mock_load.call_args[0][0]
+    # On Windows, UNC paths have \\server\share\ as a single "drive" part
+    # On POSIX, they're separate: /, server, share, schemas, pet.json
+    path_str = str(called_path)
+    assert "server" in path_str
+    assert "share" in path_str
+    assert called_path.parts[-2:] == ("schemas", "pet.json")
+
+
+def test_get_ref_body_from_url_file_local_path(mocker: MockerFixture) -> None:
+    """Test _get_ref_body_from_url handles local file:// URLs (no netloc)."""
+    parser = JsonSchemaParser("")
+    mock_load = mocker.patch(
+        "datamodel_code_generator.parser.jsonschema.load_yaml_dict_from_path",
+        return_value={"type": "string"},
+    )
+
+    result = parser._get_ref_body_from_url("file:///home/user/schemas/pet.json")
+
+    assert result == {"type": "string"}
+    mock_load.assert_called_once()
+    called_path = mock_load.call_args[0][0]
+    assert called_path.parts[-4:] == ("home", "user", "schemas", "pet.json")

tests/test_reference.py

@@ -6,6 +6,7 @@
 
 import pytest
 
+from datamodel_code_generator.http import join_url
 from datamodel_code_generator.reference import ModelResolver, get_relative_path, is_url
 
 
@@ -171,12 +172,15 @@ def test_resolve_ref_local_fragment_with_base_url() -> None:
 @pytest.mark.parametrize(
     ("ref", "expected"),
     [
-        # HTTP/HTTPS URLs (only supported schemes)
+        # HTTP/HTTPS URLs
         ("https://example.com/schema.json", True),
         ("http://example.com/schema.json", True),
         ("https://example.com/path/to/schema.json", True),
-        # file:// URLs - NOT recognized (fetcher only supports HTTP)
-        ("file:///home/user/schema.json", False),
+        # file:// URLs - recognized and handled via filesystem
+        ("file:///home/user/schema.json", True),
+        ("file:///C:/path/to/schema.json", True),
+        ("file://server/share/schema.json", True),
+        # file:/ (single slash) - NOT recognized as valid file URL
         ("file:/home/user/schema.json", False),
         # Other URL schemes - NOT recognized
         ("ftp://example.com/schema.json", False),
@@ -194,7 +198,7 @@ def test_resolve_ref_local_fragment_with_base_url() -> None:
     ],
 )
 def test_is_url(ref: str, expected: bool) -> None:
-    """Test is_url correctly identifies HTTP(S) URLs only."""
+    """Test is_url correctly identifies HTTP(S) and file:// URLs."""
     assert is_url(ref) == expected
 
 
@@ -207,3 +211,52 @@ def test_resolve_ref_with_root_id_differs_from_base_url() -> None:
     result = resolver.resolve_ref("../common/types.json")
 
     assert result == "https://example.com/common/types.json#"
+
+
+@pytest.mark.parametrize(
+    ("base_url", "ref", "expected"),
+    [
+        # file:// URL joining - relative refs
+        ("file:///home/user/schemas/main.json", "../common/types.json", "file:///home/user/common/types.json"),
+        ("file:///home/user/schemas/main.json", "other.json", "file:///home/user/schemas/other.json"),
+        ("file:///home/user/schemas/main.json", "./sub/schema.json", "file:///home/user/schemas/sub/schema.json"),
+        # file:// URL joining - absolute file:// refs
+        ("file:///home/user/schemas/main.json", "file:///other/schema.json", "file:///other/schema.json"),
+        # file:// URL joining - absolute path refs (starts with /)
+        ("file:///home/user/schemas/main.json", "/absolute/path.json", "file:///absolute/path.json"),
+        ("file://server/share/main.json", "/absolute/path.json", "file://server/absolute/path.json"),
+        # Windows-style file:// URLs
+        ("file:///C:/schemas/main.json", "../common/types.json", "file:///C:/common/types.json"),
+        # UNC file:// URLs
+        ("file://server/share/main.json", "../common/types.json", "file://server/share/common/types.json"),
+        ("file://server/share/main.json", "child.json", "file://server/share/child.json"),
+        # Fragment handling
+        (
+            "file:///home/user/schemas/main.json",
+            "other.json#/definitions/Foo",
+            "file:///home/user/schemas/other.json#/definitions/Foo",
+        ),
+        (
+            "file:///home/user/schemas/main.json",
+            "#/definitions/Bar",
+            "file:///home/user/schemas/main.json#/definitions/Bar",
+        ),
+        # Multiple .. traversal - stops at root for non-UNC
+        ("file:///a/b/main.json", "../../../other.json", "file:///other.json"),
+        # Multiple .. traversal - stops at share level for UNC (min_depth=1)
+        ("file://server/share/a/b/main.json", "../../../../other.json", "file://server/share/other.json"),
+        # Empty and dot segments
+        ("file:///home/user/schemas/main.json", "./", "file:///home/user/schemas/"),
+        ("file:///home/user/schemas/main.json", "a//b/./c.json", "file:///home/user/schemas/a/b/c.json"),
+        # Fragment-only ref without fragment content (just #)
+        ("file:///home/user/schemas/main.json", "#", "file:///home/user/schemas/main.json#"),
+        # Empty ref (keeps base URL unchanged)
+        ("file:///home/user/schemas/main.json", "", "file:///home/user/schemas/main.json"),
+        # Root directory base URL (triggers empty base_segments branch)
+        ("file:///", "schema.json", "file:///schema.json"),
+        ("file:///main.json", "../other.json", "file:///other.json"),
+    ],
+)
+def test_join_url_file_scheme(base_url: str, ref: str, expected: str) -> None:
+    """Test join_url correctly handles file:// URLs."""
+    assert join_url(base_url, ref) == expected