19739786445

Committed 27 Nov 2025 02:37PM UTC coverage: 83.311% (-3.5%) from 86.838%

Build # 19739786445

Build Type

Pull #9

github

Committed by

web-flow

Commit Message

Merge 494573424 into 04717a691

Pull Request Pull Request #9: Release/2.0.1

Run Details

340 of 414 new or added lines in 10 files covered. (82.13%)

7 existing lines in 2 files now uncovered.

614 of 737 relevant lines covered (83.31%)

2.5 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.75

/restcodegen/generator/spec/loader.py

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from restcodegen.generator.log import LOGGER
from restcodegen.generator.spec.fetcher import FetchSettings, SpecFetcher, SpecFetchError
from restcodegen.generator.spec.normalizer import SpecNormalizer
from restcodegen.generator.utils import is_url, name_to_snake


class SpecLoader:
    BASE_PATH = Path.cwd() / "clients" / "http"

    def __init__(
        self,
        spec: str,
        service_name: str,
        *,
        fetcher: SpecFetcher | None = None,
        normalizer: SpecNormalizer | None = None,
        fetch_settings: FetchSettings | None = None,
    ) -> None:
        self.spec_path = spec
        self.service_name = service_name
        self.cache_spec_dir = self.BASE_PATH / "schemas"

        if not self.cache_spec_dir.exists():
            self.cache_spec_dir.mkdir(parents=True, exist_ok=True)

        self.cache_spec_path = self.cache_spec_dir / f"{name_to_snake(self.service_name)}.json"
        self._fetcher = fetcher or SpecFetcher(settings=fetch_settings)
        self._normalizer = normalizer or SpecNormalizer()

    def _get_spec_by_url(self) -> dict[str, Any] | None:
        if not is_url(self.spec_path):
            return None

        try:
            spec = self._fetcher.fetch(self.spec_path)
        except SpecFetchError as exc:
            LOGGER.warning("OpenAPI spec not available by url %s: %s", self.spec_path, exc)
            return None

        self._write_cache(spec)
        return spec

    def _get_spec_from_cache(self) -> dict[str, Any]:
        try:
            with open(self.cache_spec_path) as f:
                spec = json.loads(f.read())
                self.spec_path = self.cache_spec_path  # type: ignore
                LOGGER.warning("OpenAPI spec loaded from cache: %s", self.spec_path)
                return spec
        except FileNotFoundError as e:
            raise FileNotFoundError(
                f"OpenAPI spec not available from url: {self.spec_path}, and not found in cache"
            ) from e

    def _get_spec_by_path(self) -> dict[str, Any] | None:
        try:
            with open(self.spec_path) as f:
                spec = json.loads(f.read())
        except FileNotFoundError:
            LOGGER.warning("OpenAPI spec not found from local path: %s", self.spec_path)
            return None
        else:
            return spec

    def _write_cache(self, spec: dict[str, Any]) -> None:
        try:
            with open(self.cache_spec_path, "w", encoding="utf-8") as f:
                f.write(json.dumps(spec, indent=4, ensure_ascii=False))
        except OSError as exc:
            LOGGER.warning("Unable to write cache file %s: %s", self.cache_spec_path, exc)

    def open(self) -> dict[str, Any]:
        spec: dict[str, Any] | None = self._get_spec_by_url()
        if spec is None:
            spec = self._get_spec_by_path()
        if spec is None:
            spec = self._get_spec_from_cache()

        return self._normalizer.normalize(spec)

1	from __future__ import annotations	3✔
2
3	import json	3✔
4	from pathlib import Path	3✔
5	from typing import Any	3✔
6
7	from restcodegen.generator.log import LOGGER	3✔
8	from restcodegen.generator.spec.fetcher import FetchSettings, SpecFetcher, SpecFetchError	3✔
9	from restcodegen.generator.spec.normalizer import SpecNormalizer	3✔
10	from restcodegen.generator.utils import is_url, name_to_snake	3✔
11
12
13	class SpecLoader:	3✔
14	BASE_PATH = Path.cwd() / "clients" / "http"	3✔
15
16	def __init__(	3✔
17	self,
18	spec: str,
19	service_name: str,
20	*,
21	fetcher: SpecFetcher \| None = None,
22	normalizer: SpecNormalizer \| None = None,
23	fetch_settings: FetchSettings \| None = None,
24	) -> None:
25	self.spec_path = spec	3✔
26	self.service_name = service_name	3✔
27	self.cache_spec_dir = self.BASE_PATH / "schemas"	3✔
28
29	if not self.cache_spec_dir.exists():	3✔
30	self.cache_spec_dir.mkdir(parents=True, exist_ok=True)	3✔
31
32	self.cache_spec_path = self.cache_spec_dir / f"{name_to_snake(self.service_name)}.json"	3✔
33	self._fetcher = fetcher or SpecFetcher(settings=fetch_settings)	3✔
34	self._normalizer = normalizer or SpecNormalizer()	3✔
35
36	def _get_spec_by_url(self) -> dict[str, Any] \| None:	3✔
37	if not is_url(self.spec_path):	3✔
38	return None	3✔
39
40	try:	3✔
41	spec = self._fetcher.fetch(self.spec_path)	3✔
NEW 42	except SpecFetchError as exc:	×
NEW 43	LOGGER.warning("OpenAPI spec not available by url %s: %s", self.spec_path, exc)	×
NEW 44	return None	×
45
46	self._write_cache(spec)	3✔
47	return spec	3✔
48
49	def _get_spec_from_cache(self) -> dict[str, Any]:	3✔
50	try:	3✔
51	with open(self.cache_spec_path) as f:	3✔
NEW 52	spec = json.loads(f.read())	×
NEW 53	self.spec_path = self.cache_spec_path # type: ignore	×
NEW 54	LOGGER.warning("OpenAPI spec loaded from cache: %s", self.spec_path)	×
NEW 55	return spec	×
56	except FileNotFoundError as e:	3✔
57	raise FileNotFoundError(	3✔
58	f"OpenAPI spec not available from url: {self.spec_path}, and not found in cache"
59	) from e
60
61	def _get_spec_by_path(self) -> dict[str, Any] \| None:	3✔
62	try:	3✔
63	with open(self.spec_path) as f:	3✔
64	spec = json.loads(f.read())	3✔
65	except FileNotFoundError:	3✔
66	LOGGER.warning("OpenAPI spec not found from local path: %s", self.spec_path)	3✔
67	return None	3✔
68	else:
69	return spec	3✔
70
71	def _write_cache(self, spec: dict[str, Any]) -> None:	3✔
72	try:	3✔
73	with open(self.cache_spec_path, "w", encoding="utf-8") as f:	3✔
74	f.write(json.dumps(spec, indent=4, ensure_ascii=False))	3✔
NEW 75	except OSError as exc:	×
NEW 76	LOGGER.warning("Unable to write cache file %s: %s", self.cache_spec_path, exc)	×
77
78	def open(self) -> dict[str, Any]:	3✔
79	spec: dict[str, Any] \| None = self._get_spec_by_url()	3✔
80	if spec is None:	3✔
81	spec = self._get_spec_by_path()	3✔
82	if spec is None:	3✔
83	spec = self._get_spec_from_cache()	3✔
84
85	return self._normalizer.normalize(spec)	3✔

ValeriyMenshikov / restcodegen / 19739786445

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous