19719535838

Committed 26 Nov 2025 10:47PM UTC coverage: 86.838% (+23.4%) from 63.397%

Build # 19719535838

Build Type

Pull #8

github

Committed by

web-flow

Commit Message

Merge 4c64980c9 into 749a61074

Pull Request Pull Request #8: Release/2.0.0

Coverage Stats

36 of 38 new or added lines in 5 files covered. (94.74%)

541 of 623 relevant lines covered (86.84%)

2.61 hits per line

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

86.04

/restcodegen/generator/parser.py

import re
from pathlib import Path
from typing import Any

from pydantic import BaseModel, Field, HttpUrl

from restcodegen.generator.log import LOGGER
from restcodegen.generator.parameters import (
    BaseParameter,
    ParameterType,
)
from restcodegen.generator.spec_loader import SpecLoader
from restcodegen.generator.utils import (
    name_to_snake,
    rename_python_builtins,
    snake_to_camel,
)

TYPE_MAP = {
    "integer": "int",
    "number": "float",
    "string": "str",
    "boolean": "bool",
    "array": "list",
    "anyof": "str",
    "none": "Any",
    "object": "Any",
}

DEFAULT_HEADER_VALUE_MAP = {"int": 0, "float": 0.0, "str": "", "bool": True}


class Handler(BaseModel):
    path: str = Field(...)
    method: str = Field(...)
    tags: list = Field(...)
    summary: str | None = Field(None)
    operation_id: str | None = Field(None)
    path_parameters: list[BaseParameter] | None = Field(None)
    query_parameters: list[BaseParameter] | None = Field(None)
    headers: list[BaseParameter] | None = Field(None)
    request_body: str | None = Field(None)
    responses: dict | None = Field(None)

    def get(self, key: str, default: Any = None) -> Any:
        return getattr(self, key, default)


class OpenApiSpec(BaseModel):
    service_name: str = Field(...)
    version: str = Field(...)
    title: str = Field(...)
    description: str = Field(...)
    openapi_version: str = Field(...)
    handlers: list[Handler]
    request_models: set[str] = set()
    response_models: set[str] = set()
    api_tags: set[str] = set()
    all_tags: set[str] = set()


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

    def __init__(
        self,
        openapi_spec: str | HttpUrl,
        service_name: str,
        selected_tags: list[str] | None = None,
    ) -> None:
        self._spec_path = str(openapi_spec)

        self._service_name = service_name
        self.openapi_spec: dict = SpecLoader(self._spec_path, self._service_name).open()
        self.version: str = ""
        self.description: str = ""
        self.openapi_version: str = ""
        self.handlers: list[Handler] = []
        self._selected_tags: set[str] = set(selected_tags) if selected_tags else set()
        self.all_tags: set[str] = set()
        self.parse()

    @property
    def apis(self) -> set[str]:
        result_tags = set()
        for tag in self._selected_tags:
            if tag not in self.all_tags:
                LOGGER.warning(f"Tag {tag} not found in openapi spec")
            else:
                result_tags.add(tag)

        if not result_tags and self._selected_tags:
            LOGGER.warning("Tags not found in openapi spec, used default tags")
            return self.all_tags
        elif not result_tags and not self._selected_tags:
            return self.all_tags

        return result_tags

    @property
    def service_name(self) -> str:
        return self._service_name

    @property
    def client_type(self) -> str:
        return "http"

    def _get_request_body(self, request_body: dict | list) -> str | None:
        if isinstance(request_body, list):
            for parameter in request_body:
                if parameter.get("in") == "body":
                    schema = parameter.get("schema", {}).get("$ref", None)
                    if schema:
                        model_path = snake_to_camel(schema.split("/")[-1])
                        model_name = model_path[0].upper() + model_path[1:]
                        return model_name
        else:
            for content_type in request_body.get("content", {}).keys():
                schema = request_body.get("content", {}).get(content_type, {}).get("schema", {}).get("$ref", None)
                if schema:
                    model_name = snake_to_camel(schema.split("/")[-1])
                    return model_name
        return None

    def _get_response_body(self, response_body: dict) -> dict:
        responses: dict = {}
        if response_body:
            if self.openapi_version.startswith("3."):
                for status_code in response_body:
                    for content_type in response_body.get(status_code, {}).get("content", {}).keys():
                        schema = (
                            response_body.get(status_code, {})
                            .get("content", {})
                            .get(content_type, {})
                            .get("schema", {})
                            .get("$ref", None)
                        )
                        model_name = schema.split("/")[-1] if schema else None
                        if model_name:
                            model_name = snake_to_camel(model_name)
                            responses[status_code] = model_name
            elif self.openapi_version.startswith("2."):
                for status_code, response in response_body.items():
                    ref_schema = response.get("schema", {}).get("$ref")
                    result_schema = response.get("schema", {}).get("result")
                    schema = ref_schema or result_schema
                    if schema:
                        model = snake_to_camel(schema.split("/")[-1])
                        responses[status_code] = model
                        model_name = model[0].upper() + model[1:]
                        responses[status_code] = model_name
        return responses

    def _get_headers(self, parameters: list[dict]) -> list[BaseParameter]:
        params = self._get_params_with_types(parameters, param_type=ParameterType.HEADER)
        return params

    def _get_path_parameters(self, parameters: list[dict]) -> list[BaseParameter]:
        params = self._get_params_with_types(parameters, param_type=ParameterType.PATH)
        return params

    def _get_query_parameters(self, parameters: list[dict]) -> list[BaseParameter]:
        params = self._get_params_with_types(parameters, param_type=ParameterType.QUERY)
        return params

    @staticmethod
    def _get_params_with_types(parameters: list[dict], param_type: ParameterType) -> list[BaseParameter]:
        params: list[BaseParameter] = []
        if not parameters:
            return params
        for parameter in parameters:
            if parameter.get("in") == param_type:
                parameter_schema = parameter.get("schema", {})
                any_of = parameter_schema.get("anyOf")
                enum_ref = parameter_schema.get("$ref")

                parameter_type = parameter_schema.get("type")
                parameter_name = parameter.get("name")
                parameter_description = parameter.get("description", "")
                parameter_is_required = parameter.get("required", False)

                if any_of:
                    parameter_type = "anyof"
                if enum_ref:
                    parameter_type = snake_to_camel(enum_ref.split("/")[-1])

                # Пропускаем параметр, если имя не определено
                if parameter_name is None:
                    continue

                parameter_with_desc = BaseParameter(
                    name=parameter_name,
                    type_=parameter_type if enum_ref else TYPE_MAP[str(parameter_type).lower()],
                    description=parameter_description,
                    required=parameter_is_required,
                    default=DEFAULT_HEADER_VALUE_MAP.get(TYPE_MAP.get(parameter_type, "")),
                )

                params.append(parameter_with_desc)

        return params

    @staticmethod
    def _normalize_swagger_path(path: str, fix_builtins: bool = True) -> str:
        def replace_placeholder(match: re.Match) -> str:
            placeholder = match.group(0)[1:-1]
            if not placeholder:
                return ""

            return (
                f"{{{rename_python_builtins(name_to_snake(placeholder))}}}"
                if fix_builtins
                else f"{{{name_to_snake(placeholder)}}}"
            )

        normalized_path = re.sub(r"\{[^}]*\}", replace_placeholder, path)
        return normalized_path

    @staticmethod
    def _extract_path_params_from_url(path: str) -> list[BaseParameter]:
        params = []
        path_params = re.findall(r"\{([^}]+)\}", path)

        for param in path_params:
            param_name = name_to_snake(param)
            params.append(
                BaseParameter(
                    name=param_name,
                    type_="str",
                    description=f"Path parameter: {param_name}",
                    required=True,
                )
            )

        return params

    def parse(self) -> list[Handler]:
        info = self.openapi_spec.get("info", {})
        self.version = info.get("version", "1.0.0")
        self.description = info.get("description", "")
        self.openapi_version = self.openapi_spec.get("openapi", "") or self.openapi_spec.get("swagger", "")

        if self.openapi_version.startswith("2."):
            LOGGER.warning(
                "OpenAPI/Swagger version 2.0 is not supported. "
                "You may convert it to 3.0 with https://converter.swagger.io/ "
                "and set the local spec path in 'swagger' option in nuke.toml!"
            )

        paths = self.openapi_spec.get("paths", {})
        for path, methods in paths.items():
            for method, details in methods.items():
                self._process_method(path, method, details)
        return self.handlers

    def _process_method(self, path: str, method: str, details: dict) -> None:
        tags = details.get("tags", [])
        for tag in tags:
            self.all_tags.add(tag)

        summary = details.get("summary", "")
        operation_id = details.get("operationId", "")
        parameters = details.get("parameters", [])
        query_parameters = self._get_query_parameters(parameters)
        path_parameters = self._get_path_parameters(parameters)
        headers = self._get_headers(parameters)
        request_body = self._get_request_body(details.get("requestBody", details.get("parameters", {})))
        responses = self._get_response_body(details.get("responses", {}))

        if not path_parameters:
            path_parameters = self._extract_path_params_from_url(path)

        path_obj = Handler(
            path=self._normalize_swagger_path(path),
            method=method,
            tags=tags,
            summary=summary,
            operation_id=operation_id,
            query_parameters=query_parameters,
            headers=headers,
            path_parameters=path_parameters,
            request_body=request_body,
            responses=responses,
        )
        self.handlers.append(path_obj)

    def request_models(self) -> set[str]:
        models: set[str] = set()
        for handler in self.handlers:
            if handler.request_body is not None:
                models.add(handler.request_body)
        return models

    def response_models(self) -> set[str]:
        models: set[str] = set()
        for handler in self.handlers:
            if handler.responses is not None:
                models.update(handler.responses.values())
        return models

    def models_by_tag(self, tag: str) -> set[str]:
        models: set[str] = set()
        for handler in self.handlers:
            if tag in handler.tags:
                if handler.path_parameters is not None:
                    for param in handler.path_parameters:
                        param_type = param.type
                        if param_type not in TYPE_MAP.values():
                            models.add(param_type)
                if handler.query_parameters is not None:
                    for query_param in handler.query_parameters:
                        query_param_type = query_param.type
                        if query_param_type not in TYPE_MAP.values():
                            models.add(query_param_type)
                if handler.headers is not None:
                    for header_param in handler.headers:
                        header_param_type = header_param.type
                        if header_param_type not in TYPE_MAP.values():
                            models.add(header_param_type)
                if handler.request_body is not None:
                    models.add(handler.request_body)
                if handler.responses is not None:
                    models.update(handler.responses.values())
        return models

    def handlers_by_tag(self, tag: str) -> list[Handler]:
        return [h for h in self.handlers if tag in h.tags]

    def handlers_by_method(self, method: str) -> list[Handler]:
        return [h for h in self.handlers if h.method == method]

    def handler_by_path(self, path: str) -> list[Handler]:
        return [h for h in self.handlers if h.path == path]

1	import re	3✔
2	from pathlib import Path	3✔
3	from typing import Any	3✔
4
5	from pydantic import BaseModel, Field, HttpUrl	3✔
6
7	from restcodegen.generator.log import LOGGER	3✔
8	from restcodegen.generator.parameters import (	3✔
9	BaseParameter,
10	ParameterType,
11	)
12	from restcodegen.generator.spec_loader import SpecLoader	3✔
13	from restcodegen.generator.utils import (	3✔
14	name_to_snake,
15	rename_python_builtins,
16	snake_to_camel,
17	)
18
19	TYPE_MAP = {	3✔
20	"integer": "int",
21	"number": "float",
22	"string": "str",
23	"boolean": "bool",
24	"array": "list",
25	"anyof": "str",
26	"none": "Any",
27	"object": "Any",
28	}
29
30	DEFAULT_HEADER_VALUE_MAP = {"int": 0, "float": 0.0, "str": "", "bool": True}	3✔
31
32
33	class Handler(BaseModel):	3✔
34	path: str = Field(...)	3✔
35	method: str = Field(...)	3✔
36	tags: list = Field(...)	3✔
37	summary: str \| None = Field(None)	3✔
38	operation_id: str \| None = Field(None)	3✔
39	path_parameters: list[BaseParameter] \| None = Field(None)	3✔
40	query_parameters: list[BaseParameter] \| None = Field(None)	3✔
41	headers: list[BaseParameter] \| None = Field(None)	3✔
42	request_body: str \| None = Field(None)	3✔
43	responses: dict \| None = Field(None)	3✔
44
45	def get(self, key: str, default: Any = None) -> Any:	3✔
46	return getattr(self, key, default)	3✔
47
48
49	class OpenApiSpec(BaseModel):	3✔
50	service_name: str = Field(...)	3✔
51	version: str = Field(...)	3✔
52	title: str = Field(...)	3✔
53	description: str = Field(...)	3✔
54	openapi_version: str = Field(...)	3✔
55	handlers: list[Handler]	3✔
56	request_models: set[str] = set()	3✔
57	response_models: set[str] = set()	3✔
58	api_tags: set[str] = set()	3✔
59	all_tags: set[str] = set()	3✔
60
61
62	class Parser:	3✔
63	BASE_PATH = Path.cwd() / "clients" / "http"	3✔
64
65	def __init__(	3✔
66	self,
67	openapi_spec: str \| HttpUrl,
68	service_name: str,
69	selected_tags: list[str] \| None = None,
70	) -> None:
71	self._spec_path = str(openapi_spec)	3✔
72
73	self._service_name = service_name	3✔
74	self.openapi_spec: dict = SpecLoader(self._spec_path, self._service_name).open()	3✔
75	self.version: str = ""	3✔
76	self.description: str = ""	3✔
77	self.openapi_version: str = ""	3✔
78	self.handlers: list[Handler] = []	3✔
79	self._selected_tags: set[str] = set(selected_tags) if selected_tags else set()	3✔
80	self.all_tags: set[str] = set()	3✔
81	self.parse()	3✔
82
83	@property	3✔
84	def apis(self) -> set[str]:	3✔
85	result_tags = set()	3✔
86	for tag in self._selected_tags:	3✔
87	if tag not in self.all_tags:	3✔
88	LOGGER.warning(f"Tag {tag} not found in openapi spec")	3✔
89	else:
90	result_tags.add(tag)	3✔
91
92	if not result_tags and self._selected_tags:	3✔
93	LOGGER.warning("Tags not found in openapi spec, used default tags")	3✔
94	return self.all_tags	3✔
95	elif not result_tags and not self._selected_tags:	3✔
96	return self.all_tags	3✔
97
98	return result_tags	3✔
99
100	@property	3✔
101	def service_name(self) -> str:	3✔
102	return self._service_name	3✔
103
104	@property	3✔
105	def client_type(self) -> str:	3✔
106	return "http"	3✔
107
108	def _get_request_body(self, request_body: dict \| list) -> str \| None:	3✔
109	if isinstance(request_body, list):	3✔
110	for parameter in request_body:	3✔
111	if parameter.get("in") == "body":	3✔
112	schema = parameter.get("schema", {}).get("$ref", None)	×
113	if schema:	×
114	model_path = snake_to_camel(schema.split("/")[-1])	×
115	model_name = model_path[0].upper() + model_path[1:]	×
116	return model_name	×
117	else:
118	for content_type in request_body.get("content", {}).keys():	3✔
119	schema = request_body.get("content", {}).get(content_type, {}).get("schema", {}).get("$ref", None)	3✔
120	if schema:	3✔
121	model_name = snake_to_camel(schema.split("/")[-1])	3✔
122	return model_name	3✔
123	return None	3✔
124
125	def _get_response_body(self, response_body: dict) -> dict:	3✔
126	responses: dict = {}	3✔
127	if response_body:	3✔
128	if self.openapi_version.startswith("3."):	3✔
129	for status_code in response_body:	3✔
130	for content_type in response_body.get(status_code, {}).get("content", {}).keys():	3✔
131	schema = (	3✔
132	response_body.get(status_code, {})
133	.get("content", {})
134	.get(content_type, {})
135	.get("schema", {})
136	.get("$ref", None)
137	)
138	model_name = schema.split("/")[-1] if schema else None	3✔
139	if model_name:	3✔
140	model_name = snake_to_camel(model_name)	3✔
141	responses[status_code] = model_name	3✔
142	elif self.openapi_version.startswith("2."):	×
143	for status_code, response in response_body.items():	×
144	ref_schema = response.get("schema", {}).get("$ref")	×
145	result_schema = response.get("schema", {}).get("result")	×
146	schema = ref_schema or result_schema	×
147	if schema:	×
148	model = snake_to_camel(schema.split("/")[-1])	×
149	responses[status_code] = model	×
150	model_name = model[0].upper() + model[1:]	×
151	responses[status_code] = model_name	×
152	return responses	3✔
153
154	def _get_headers(self, parameters: list[dict]) -> list[BaseParameter]:	3✔
155	params = self._get_params_with_types(parameters, param_type=ParameterType.HEADER)	3✔
156	return params	3✔
157
158	def _get_path_parameters(self, parameters: list[dict]) -> list[BaseParameter]:	3✔
159	params = self._get_params_with_types(parameters, param_type=ParameterType.PATH)	3✔
160	return params	3✔
161
162	def _get_query_parameters(self, parameters: list[dict]) -> list[BaseParameter]:	3✔
163	params = self._get_params_with_types(parameters, param_type=ParameterType.QUERY)	3✔
164	return params	3✔
165
166	@staticmethod	3✔
167	def _get_params_with_types(parameters: list[dict], param_type: ParameterType) -> list[BaseParameter]:	3✔
168	params: list[BaseParameter] = []	3✔
169	if not parameters:	3✔
170	return params	3✔
171	for parameter in parameters:	3✔
172	if parameter.get("in") == param_type:	3✔
173	parameter_schema = parameter.get("schema", {})	3✔
174	any_of = parameter_schema.get("anyOf")	3✔
175	enum_ref = parameter_schema.get("$ref")	3✔
176
177	parameter_type = parameter_schema.get("type")	3✔
178	parameter_name = parameter.get("name")	3✔
179	parameter_description = parameter.get("description", "")	3✔
180	parameter_is_required = parameter.get("required", False)	3✔
181
182	if any_of:	3✔
183	parameter_type = "anyof"	×
184	if enum_ref:	3✔
NEW 185	parameter_type = snake_to_camel(enum_ref.split("/")[-1])	×
186
187	# Пропускаем параметр, если имя не определено
188	if parameter_name is None:	3✔
189	continue	×
190
191	parameter_with_desc = BaseParameter(	3✔
192	name=parameter_name,
193	type_=parameter_type if enum_ref else TYPE_MAP[str(parameter_type).lower()],
194	description=parameter_description,
195	required=parameter_is_required,
196	default=DEFAULT_HEADER_VALUE_MAP.get(TYPE_MAP.get(parameter_type, "")),
197	)
198
199	params.append(parameter_with_desc)	3✔
200
201	return params	3✔
202
203	@staticmethod	3✔
204	def _normalize_swagger_path(path: str, fix_builtins: bool = True) -> str:	3✔
205	def replace_placeholder(match: re.Match) -> str:	3✔
206	placeholder = match.group(0)[1:-1]	3✔
207	if not placeholder:	3✔
208	return ""	×
209
210	return (	3✔
211	f"{{{rename_python_builtins(name_to_snake(placeholder))}}}"
212	if fix_builtins
213	else f"{{{name_to_snake(placeholder)}}}"
214	)
215
216	normalized_path = re.sub(r"\{[^}]*\}", replace_placeholder, path)	3✔
217	return normalized_path	3✔
218
219	@staticmethod	3✔
220	def _extract_path_params_from_url(path: str) -> list[BaseParameter]:	3✔
221	params = []	3✔
222	path_params = re.findall(r"\{([^}]+)\}", path)	3✔
223
224	for param in path_params:	3✔
225	param_name = name_to_snake(param)	×
226	params.append(	×
227	BaseParameter(
228	name=param_name,
229	type_="str",
230	description=f"Path parameter: {param_name}",
231	required=True,
232	)
233	)
234
235	return params	3✔
236
237	def parse(self) -> list[Handler]:	3✔
238	info = self.openapi_spec.get("info", {})	3✔
239	self.version = info.get("version", "1.0.0")	3✔
240	self.description = info.get("description", "")	3✔
241	self.openapi_version = self.openapi_spec.get("openapi", "") or self.openapi_spec.get("swagger", "")	3✔
242
243	if self.openapi_version.startswith("2."):	3✔
244	LOGGER.warning(	×
245	"OpenAPI/Swagger version 2.0 is not supported. "
246	"You may convert it to 3.0 with https://converter.swagger.io/ "
247	"and set the local spec path in 'swagger' option in nuke.toml!"
248	)
249
250	paths = self.openapi_spec.get("paths", {})	3✔
251	for path, methods in paths.items():	3✔
252	for method, details in methods.items():	3✔
253	self._process_method(path, method, details)	3✔
254	return self.handlers	3✔
255
256	def _process_method(self, path: str, method: str, details: dict) -> None:	3✔
257	tags = details.get("tags", [])	3✔
258	for tag in tags:	3✔
259	self.all_tags.add(tag)	3✔
260
261	summary = details.get("summary", "")	3✔
262	operation_id = details.get("operationId", "")	3✔
263	parameters = details.get("parameters", [])	3✔
264	query_parameters = self._get_query_parameters(parameters)	3✔
265	path_parameters = self._get_path_parameters(parameters)	3✔
266	headers = self._get_headers(parameters)	3✔
267	request_body = self._get_request_body(details.get("requestBody", details.get("parameters", {})))	3✔
268	responses = self._get_response_body(details.get("responses", {}))	3✔
269
270	if not path_parameters:	3✔
271	path_parameters = self._extract_path_params_from_url(path)	3✔
272
273	path_obj = Handler(	3✔
274	path=self._normalize_swagger_path(path),
275	method=method,
276	tags=tags,
277	summary=summary,
278	operation_id=operation_id,
279	query_parameters=query_parameters,
280	headers=headers,
281	path_parameters=path_parameters,
282	request_body=request_body,
283	responses=responses,
284	)
285	self.handlers.append(path_obj)	3✔
286
287	def request_models(self) -> set[str]:	3✔
288	models: set[str] = set()	3✔
289	for handler in self.handlers:	3✔
290	if handler.request_body is not None:	3✔
291	models.add(handler.request_body)	3✔
292	return models	3✔
293
294	def response_models(self) -> set[str]:	3✔
295	models: set[str] = set()	3✔
296	for handler in self.handlers:	3✔
297	if handler.responses is not None:	3✔
298	models.update(handler.responses.values())	3✔
299	return models	3✔
300
301	def models_by_tag(self, tag: str) -> set[str]:	3✔
302	models: set[str] = set()	3✔
303	for handler in self.handlers:	3✔
304	if tag in handler.tags:	3✔
305	if handler.path_parameters is not None:	3✔
306	for param in handler.path_parameters:	3✔
307	param_type = param.type	×
308	if param_type not in TYPE_MAP.values():	×
309	models.add(param_type)	×
310	if handler.query_parameters is not None:	3✔
311	for query_param in handler.query_parameters:	3✔
312	query_param_type = query_param.type	3✔
313	if query_param_type not in TYPE_MAP.values():	3✔
314	models.add(query_param_type)	×
315	if handler.headers is not None:	3✔
316	for header_param in handler.headers:	3✔
317	header_param_type = header_param.type	×
318	if header_param_type not in TYPE_MAP.values():	×
319	models.add(header_param_type)	×
320	if handler.request_body is not None:	3✔
321	models.add(handler.request_body)	3✔
322	if handler.responses is not None:	3✔
323	models.update(handler.responses.values())	3✔
324	return models	3✔
325
326	def handlers_by_tag(self, tag: str) -> list[Handler]:	3✔
327	return [h for h in self.handlers if tag in h.tags]	3✔
328
329	def handlers_by_method(self, method: str) -> list[Handler]:	3✔
330	return [h for h in self.handlers if h.method == method]	×
331
332	def handler_by_path(self, path: str) -> list[Handler]:	3✔
333	return [h for h in self.handlers if h.path == path]	×

ValeriyMenshikov / restcodegen / 19719535838

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