diff --git a/src/openapi_parser/builders/common.py b/src/openapi_parser/builders/common.py index cd34983..8fdaf5e 100644 --- a/src/openapi_parser/builders/common.py +++ b/src/openapi_parser/builders/common.py @@ -35,14 +35,21 @@ def cast_value( type_cast_func: Callable[..., Any] | None, ) -> Any: try: - return type_cast_func(value) if type_cast_func is not None else value + if type_cast_func is not None: + return type_cast_func(value) + + return value except ValueError: raise ParserError( - f"Invalid value for '{name}' property, got '{value}'" + f"Invalid value for '{name}' property, got '{value}'", ) from None custom_attrs = { - attr_name: cast_value(attr_info.name, data[attr_info.name], attr_info.cast) + attr_name: cast_value( + attr_info.name, + data[attr_info.name], + attr_info.cast, + ) for attr_name, attr_info in attrs_map.items() if data.get(attr_info.name) is not None } diff --git a/src/openapi_parser/builders/content.py b/src/openapi_parser/builders/content.py index 4c0d92d..277c913 100644 --- a/src/openapi_parser/builders/content.py +++ b/src/openapi_parser/builders/content.py @@ -6,6 +6,7 @@ from openapi_parser.builders.encoding import EncodingBuilder from openapi_parser.builders.schema import SchemaFactory from openapi_parser.enumeration import ContentType +from openapi_parser.logging import log_ctx from openapi_parser.loose_types import LooseContentType from openapi_parser.specification import Content @@ -38,13 +39,13 @@ def __init__( self._encoding_builder = encoding_builder self._strict_enum = strict_enum - def build_list(self, data: dict[str, Any]) -> list[Content]: + def build_list( + self, + data: dict[str, Any], + ) -> list[Content]: """Build a list of content objects from a dict of media types.""" return [ - self._create_content( - content_type, - content_value, - ) + self._create_content(content_type, content_value) for content_type, content_value in data.items() ] @@ -53,22 +54,23 @@ def _create_content( content_type: str, content_value: dict[str, Any], ) -> Content: - logger.debug(f"Content building [type={content_type}]") + with log_ctx("content", content_type): + logger.debug(f"Content building [type={content_type}]") - ContentTypeCls: ContentTypeType = ( - ContentType if self._strict_enum else LooseContentType - ) + ContentTypeCls: ContentTypeType = ( + ContentType if self._strict_enum else LooseContentType + ) - encoding = ( - self._encoding_builder.build_dict(content_value["encoding"]) - if content_value.get("encoding") - else None - ) + encoding = ( + self._encoding_builder.build_dict(content_value["encoding"]) + if content_value.get("encoding") + else None + ) - return Content( - type=ContentTypeCls(content_type), - schema=self._schema_factory.create(content_value.get("schema", {})), - example=content_value.get("example"), - examples=content_value.get("examples", {}), - encoding=encoding, - ) + return Content( + type=ContentTypeCls(content_type), + schema=self._schema_factory.create(content_value.get("schema", {})), + example=content_value.get("example"), + examples=content_value.get("examples", {}), + encoding=encoding, + ) diff --git a/src/openapi_parser/builders/encoding.py b/src/openapi_parser/builders/encoding.py index 9a7a36b..8b3cfc0 100644 --- a/src/openapi_parser/builders/encoding.py +++ b/src/openapi_parser/builders/encoding.py @@ -9,6 +9,7 @@ extract_typed_props, ) from openapi_parser.builders.header import HeaderBuilder +from openapi_parser.logging import log_ctx from openapi_parser.specification import Encoding logger = logging.getLogger(__name__) @@ -27,12 +28,18 @@ def __init__(self, header_builder: HeaderBuilder) -> None: """ self._header_builder = header_builder - def build_dict(self, data: dict[str, dict[str, Any]]) -> dict[str, Encoding]: + def build_dict( + self, + data: dict[str, dict[str, Any]], + ) -> dict[str, Encoding]: """Build a dict of encodings from a dict of raw encoding definitions.""" - return { - property_name: self._build(encoding_data) - for property_name, encoding_data in data.items() - } + result: dict[str, Encoding] = {} + + for property_name, encoding_data in data.items(): + with log_ctx("encoding", property_name): + result[property_name] = self._build(encoding_data) + + return result def _build(self, data: dict[str, Any]) -> Encoding: logger.debug("Encoding building") diff --git a/src/openapi_parser/builders/header.py b/src/openapi_parser/builders/header.py index 93ee0ce..f64ef26 100644 --- a/src/openapi_parser/builders/header.py +++ b/src/openapi_parser/builders/header.py @@ -9,6 +9,7 @@ extract_typed_props, ) from openapi_parser.builders.schema import SchemaFactory +from openapi_parser.logging import log_ctx from openapi_parser.specification import Header logger = logging.getLogger(__name__) @@ -27,29 +28,41 @@ def __init__(self, schema_factory: SchemaFactory) -> None: """ self._schema_factory = schema_factory - def build_list(self, data: dict[str, Any]) -> list[Header]: + def build_list( + self, + data: dict[str, Any], + ) -> list[Header]: """Build a list of headers from a dict of header definitions.""" return [ self._build(header_name, header_value) for header_name, header_value in data.items() ] - def _build(self, name: str, data: dict[str, Any]) -> Header: - logger.debug(f"Header parsing: {name}") + def _build( + self, + name: str, + data: dict[str, Any], + ) -> Header: + with log_ctx("headers", name): + logger.debug(f"Header parsing: {name}") - attrs_map = { - "schema": PropertyMeta(name="schema", cast=self._schema_factory.create), - "description": PropertyMeta(name="description", cast=str), - "deprecated": PropertyMeta(name="deprecated", cast=bool), - "required": PropertyMeta(name="required", cast=bool), - } + attrs_map = { + "description": PropertyMeta(name="description", cast=str), + "deprecated": PropertyMeta(name="deprecated", cast=bool), + "required": PropertyMeta(name="required", cast=bool), + } - attrs = extract_typed_props(data, attrs_map) + attrs = extract_typed_props(data, attrs_map) - attrs["name"] = name - attrs["extensions"] = extract_extension_attributes(data) + if data.get("schema") is not None: + attrs["schema"] = self._schema_factory.create(data["schema"]) - if attrs["extensions"]: - logger.debug(f"Extracted custom properties [{attrs['extensions'].keys()}]") + attrs["name"] = name + attrs["extensions"] = extract_extension_attributes(data) - return Header(**attrs) + if attrs["extensions"]: + logger.debug( + f"Extracted custom properties [{attrs['extensions'].keys()}]" + ) + + return Header(**attrs) diff --git a/src/openapi_parser/builders/info.py b/src/openapi_parser/builders/info.py index 0ed9a55..5c41d91 100644 --- a/src/openapi_parser/builders/info.py +++ b/src/openapi_parser/builders/info.py @@ -9,6 +9,7 @@ extract_typed_props, ) from openapi_parser.errors import ParserError +from openapi_parser.logging import log_ctx from openapi_parser.specification import Contact, Info, License logger = logging.getLogger(__name__) @@ -19,29 +20,34 @@ class InfoBuilder: def build(self, data: dict[str, Any]) -> Info: """Build an Info object from a raw dict.""" - title = data.get("title") - - if title is None: - raise ParserError("Info section is missing required 'title' property") - - logger.debug(f"Info section parsing [title={title}]") - - attrs_map = { - "title": PropertyMeta(name="title", cast=str), - "version": PropertyMeta(name="version", cast=str), - "description": PropertyMeta(name="description", cast=str), - "terms_of_service": PropertyMeta(name="termsOfService", cast=str), - "license": PropertyMeta(name="license", cast=self._create_license), - "contact": PropertyMeta(name="contact", cast=self._create_contact), - } - - attrs = extract_typed_props(data, attrs_map) - attrs["extensions"] = extract_extension_attributes(data) - - if attrs["extensions"]: - logger.debug(f"Extracted custom properties [{attrs['extensions'].keys()}]") - - return Info(**attrs) + with log_ctx("info"): + title = data.get("title") + + if title is None: + raise ParserError( + "Info section is missing required 'title' property", + ) + + logger.debug(f"Info section parsing [title={title}]") + + attrs_map = { + "title": PropertyMeta(name="title", cast=str), + "version": PropertyMeta(name="version", cast=str), + "description": PropertyMeta(name="description", cast=str), + "terms_of_service": PropertyMeta(name="termsOfService", cast=str), + "license": PropertyMeta(name="license", cast=self._create_license), + "contact": PropertyMeta(name="contact", cast=self._create_contact), + } + + attrs = extract_typed_props(data, attrs_map) + attrs["extensions"] = extract_extension_attributes(data) + + if attrs["extensions"]: + logger.debug( + f"Extracted custom properties [{attrs['extensions'].keys()}]" + ) + + return Info(**attrs) @staticmethod def _create_license(data: dict[str, Any]) -> License: diff --git a/src/openapi_parser/builders/link.py b/src/openapi_parser/builders/link.py index bd54c0a..ae7d7ee 100644 --- a/src/openapi_parser/builders/link.py +++ b/src/openapi_parser/builders/link.py @@ -8,6 +8,7 @@ extract_extension_attributes, extract_typed_props, ) +from openapi_parser.logging import log_ctx from openapi_parser.specification import Link, Server logger = logging.getLogger(__name__) @@ -24,11 +25,18 @@ def build_server(value: dict[str, Any]) -> Server: class LinkBuilder: """Builds link objects from raw specification data.""" - def build_dict(self, data: dict[str, dict[str, Any]]) -> dict[str, Link]: + def build_dict( + self, + data: dict[str, dict[str, Any]], + ) -> dict[str, Link]: """Build a dict of links from a dict of raw link definitions.""" - return { - link_name: self._build(link_data) for link_name, link_data in data.items() - } + result: dict[str, Link] = {} + + for link_name, link_data in data.items(): + with log_ctx("links", link_name): + result[link_name] = self._build(link_data) + + return result def _build(self, data: dict[str, Any]) -> Link: logger.debug("Link building") diff --git a/src/openapi_parser/builders/oauth_flow.py b/src/openapi_parser/builders/oauth_flow.py index 25ce005..cc7dfdb 100644 --- a/src/openapi_parser/builders/oauth_flow.py +++ b/src/openapi_parser/builders/oauth_flow.py @@ -18,7 +18,9 @@ class OAuthFlowBuilder: """Builds OAuth flow collections from raw specification data.""" @staticmethod - def build_collection(data: dict[str, Any]) -> dict[OAuthFlowType, OAuthFlow]: + def build_collection( + data: dict[str, Any], + ) -> dict[OAuthFlowType, OAuthFlow]: """Build a dict of OAuthFlow objects from a raw dict.""" logger.debug(f"Parsing OAuth items collection: {data.keys()}") @@ -45,7 +47,8 @@ def build_collection(data: dict[str, Any]) -> dict[OAuthFlowType, OAuthFlow]: for extension in extensions: result_oauth_dict[cast(OAuthFlowType, extension)] = cast( - OAuthFlow, extensions[extension] + OAuthFlow, + extensions[extension], ) return result_oauth_dict diff --git a/src/openapi_parser/builders/operation.py b/src/openapi_parser/builders/operation.py index 737a43e..4ea4cff 100644 --- a/src/openapi_parser/builders/operation.py +++ b/src/openapi_parser/builders/operation.py @@ -13,6 +13,7 @@ from openapi_parser.builders.request import RequestBuilder from openapi_parser.builders.response import ResponseBuilder from openapi_parser.enumeration import OperationMethod +from openapi_parser.logging import log_ctx from openapi_parser.specification import Operation, Response logger = logging.getLogger(__name__) @@ -46,46 +47,59 @@ def __init__( self._request_builder = request_builder self._parameter_builder = parameter_builder - def build(self, method: OperationMethod, data: dict[str, Any]) -> Operation: + def build( + self, + method: OperationMethod, + data: dict[str, Any], + ) -> Operation: """Build an Operation from a method and raw data dict.""" - logger.info( - f"Operation item parsing [method={method.value}, id={data.get('operationId')}]", - ) - - attrs_map = { - "responses": PropertyMeta(name="responses", cast=self._get_response_list), - "summary": PropertyMeta(name="summary", cast=str), - "description": PropertyMeta(name="description", cast=str), - "operation_id": PropertyMeta(name="operationId", cast=str), - "external_docs": PropertyMeta( - name="externalDocs", - cast=self._external_doc_builder.build, - ), - "request_body": PropertyMeta( - name="requestBody", - cast=self._request_builder.build, - ), - "deprecated": PropertyMeta(name="deprecated", cast=bool), - "parameters": PropertyMeta( - name="parameters", - cast=self._parameter_builder.build_list, - ), - "tags": PropertyMeta(name="tags", cast=list), - "security": PropertyMeta(name="security", cast=None), - "callbacks": PropertyMeta(name="callbacks", cast=None), - } - - attrs = extract_typed_props(data, attrs_map) - attrs["extensions"] = extract_extension_attributes(data) - - if attrs["extensions"]: - logger.debug(f"Extracted custom properties [{attrs['extensions'].keys()}]") - - attrs["method"] = method - - return Operation(**attrs) - - def _get_response_list(self, data: dict[str, dict[str, Any]]) -> list[Response]: + with log_ctx(method.value): + logger.info( + f"Operation item parsing [method={method.value}, id={data.get('operationId')}]", + ) + + attrs_map = { + "summary": PropertyMeta(name="summary", cast=str), + "description": PropertyMeta(name="description", cast=str), + "operation_id": PropertyMeta(name="operationId", cast=str), + "external_docs": PropertyMeta( + name="externalDocs", + cast=self._external_doc_builder.build, + ), + "request_body": PropertyMeta( + name="requestBody", + cast=self._request_builder.build, + ), + "deprecated": PropertyMeta(name="deprecated", cast=bool), + "parameters": PropertyMeta( + name="parameters", + cast=self._parameter_builder.build_list, + ), + "tags": PropertyMeta(name="tags", cast=list), + "security": PropertyMeta(name="security", cast=None), + "callbacks": PropertyMeta(name="callbacks", cast=None), + } + + attrs = extract_typed_props(data, attrs_map) + + if data.get("responses") is not None: + attrs["responses"] = self._get_response_list(data["responses"]) + + attrs["extensions"] = extract_extension_attributes(data) + + if attrs["extensions"]: + logger.debug( + f"Extracted custom properties [{attrs['extensions'].keys()}]" + ) + + attrs["method"] = method + + return Operation(**attrs) + + def _get_response_list( + self, + data: dict[str, dict[str, Any]], + ) -> list[Response]: return [ self._response_builder.build(http_code, response) for http_code, response in data.items() diff --git a/src/openapi_parser/builders/parameter.py b/src/openapi_parser/builders/parameter.py index c60c69f..cc599ca 100644 --- a/src/openapi_parser/builders/parameter.py +++ b/src/openapi_parser/builders/parameter.py @@ -18,6 +18,7 @@ QueryParameterStyle, ) from openapi_parser.errors import ParserError +from openapi_parser.logging import log_ctx from openapi_parser.specification import Parameter logger = logging.getLogger(__name__) @@ -57,49 +58,59 @@ def __init__( self._schema_factory = schema_factory self._content_builder = content_builder - def build_list(self, parameters: list[dict[str, Any]]) -> list[Parameter]: + def build_list( + self, + parameters: list[dict[str, Any]], + ) -> list[Parameter]: """Build a list of parameters from a list of raw dicts.""" return [self.build(parameter) for parameter in parameters] def build(self, data: dict[str, Any]) -> Parameter: """Build a Parameter from a raw dict.""" - parameter_name = data.get("name") - - if parameter_name is None: - raise ParserError("Parameter is missing required 'name' property") - - logger.debug(f"Parameter parsing [name={parameter_name}]") - - attrs_map = { - "name": PropertyMeta(name="name", cast=str), - "location": PropertyMeta(name="in", cast=ParameterLocation), - "required": PropertyMeta(name="required", cast=bool), - "schema": PropertyMeta(name="schema", cast=self._schema_factory.create), - "content": PropertyMeta( - name="content", - cast=self._content_builder.build_list, - ), - "description": PropertyMeta(name="description", cast=str), - "example": PropertyMeta(name="example", cast=None), - "examples": PropertyMeta(name="examples", cast=dict), - "deprecated": PropertyMeta(name="deprecated", cast=bool), - "explode": PropertyMeta(name="explode", cast=bool), - "allow_reserved": PropertyMeta(name="allowReserved", cast=bool), - } - - attrs = extract_typed_props(data, attrs_map) - - if data.get("style"): - attrs["style"] = style_to_enum_map[attrs["location"]](data["style"]) - else: - attrs["style"] = default_styles_by_location[attrs["location"]] - - if not attrs.get("explode") and attrs["style"].value == "form": - attrs["explode"] = True - - attrs["extensions"] = extract_extension_attributes(data) - - if attrs["extensions"]: - logger.debug(f"Extracted custom properties [{attrs['extensions'].keys()}]") - - return Parameter(**attrs) + with log_ctx("parameters"): + parameter_name = data.get("name") + + if parameter_name is None: + raise ParserError( + "Parameter is missing required 'name' property", + ) + + with log_ctx(parameter_name): + logger.debug(f"Parameter parsing [name={parameter_name}]") + + attrs_map = { + "name": PropertyMeta(name="name", cast=str), + "location": PropertyMeta(name="in", cast=ParameterLocation), + "required": PropertyMeta(name="required", cast=bool), + "description": PropertyMeta(name="description", cast=str), + "example": PropertyMeta(name="example", cast=None), + "examples": PropertyMeta(name="examples", cast=dict), + "deprecated": PropertyMeta(name="deprecated", cast=bool), + "explode": PropertyMeta(name="explode", cast=bool), + "allow_reserved": PropertyMeta(name="allowReserved", cast=bool), + } + + attrs = extract_typed_props(data, attrs_map) + + if data.get("schema") is not None: + attrs["schema"] = self._schema_factory.create(data["schema"]) + + if data.get("content") is not None: + attrs["content"] = self._content_builder.build_list(data["content"]) + + if data.get("style"): + attrs["style"] = style_to_enum_map[attrs["location"]](data["style"]) + else: + attrs["style"] = default_styles_by_location[attrs["location"]] + + if not attrs.get("explode") and attrs["style"].value == "form": + attrs["explode"] = True + + attrs["extensions"] = extract_extension_attributes(data) + + if attrs["extensions"]: + logger.debug( + f"Extracted custom properties [{attrs['extensions'].keys()}]" + ) + + return Parameter(**attrs) diff --git a/src/openapi_parser/builders/path.py b/src/openapi_parser/builders/path.py index 0fe10c1..3f128b2 100644 --- a/src/openapi_parser/builders/path.py +++ b/src/openapi_parser/builders/path.py @@ -11,6 +11,7 @@ from openapi_parser.builders.operation import OperationBuilder from openapi_parser.builders.parameter import ParameterBuilder from openapi_parser.enumeration import OperationMethod +from openapi_parser.logging import log_ctx from openapi_parser.specification import Path logger = logging.getLogger(__name__) @@ -36,40 +37,49 @@ def __init__( self._operation_builder = operation_builder self._parameter_builder = parameter_builder - def build_list(self, data: dict[str, dict[str, Any]]) -> list[Path]: + def build_list( + self, + data: dict[str, dict[str, Any]], + ) -> list[Path]: """Build a list of paths from a raw dict of path definitions.""" return [self._build_path(url, path) for url, path in data.items()] def _build_path(self, url: str, data: dict[str, Any]) -> Path: - logger.info(f"Path item parsing [url={url}]") - - attrs_map = { - "summary": PropertyMeta(name="summary", cast=str), - "description": PropertyMeta(name="description", cast=str), - "parameters": PropertyMeta( - name="parameters", - cast=self._parameter_builder.build_list, - ), - } - - attrs = extract_typed_props(data, attrs_map) - - attrs["url"] = url - - attrs["operations"] = [ - self._operation_builder.build(method, data[method.value]) - for method in OperationMethod - if method.value in data - ] - - if attrs.get("parameters"): - for operation in attrs["operations"]: - merged = operation.parameters + attrs["parameters"] - object.__setattr__(operation, "parameters", merged) - - attrs["extensions"] = extract_extension_attributes(data) - - if attrs["extensions"]: - logger.debug(f"Extracted custom properties [{attrs['extensions'].keys()}]") - - return Path(**attrs) + with log_ctx("paths", url): + logger.info(f"Path item parsing [url={url}]") + + attrs_map = { + "summary": PropertyMeta(name="summary", cast=str), + "description": PropertyMeta(name="description", cast=str), + "parameters": PropertyMeta( + name="parameters", + cast=self._parameter_builder.build_list, + ), + } + + attrs = extract_typed_props(data, attrs_map) + + attrs["url"] = url + + attrs["operations"] = [ + self._operation_builder.build( + method, + data[method.value], + ) + for method in OperationMethod + if method.value in data + ] + + if attrs.get("parameters"): + for operation in attrs["operations"]: + merged = operation.parameters + attrs["parameters"] + object.__setattr__(operation, "parameters", merged) + + attrs["extensions"] = extract_extension_attributes(data) + + if attrs["extensions"]: + logger.debug( + f"Extracted custom properties [{attrs['extensions'].keys()}]" + ) + + return Path(**attrs) diff --git a/src/openapi_parser/builders/request.py b/src/openapi_parser/builders/request.py index a668fa6..fe1ee7d 100644 --- a/src/openapi_parser/builders/request.py +++ b/src/openapi_parser/builders/request.py @@ -3,8 +3,12 @@ import logging from typing import Any -from openapi_parser.builders.common import PropertyMeta, extract_typed_props +from openapi_parser.builders.common import ( + PropertyMeta, + extract_typed_props, +) from openapi_parser.builders.content import ContentBuilder +from openapi_parser.logging import log_ctx from openapi_parser.specification import RequestBody logger = logging.getLogger(__name__) @@ -25,17 +29,18 @@ def __init__(self, content_builder: ContentBuilder) -> None: def build(self, data: dict[str, Any]) -> RequestBody: """Build a RequestBody from a raw dict.""" - logger.debug("Request building") + with log_ctx("requestBody"): + logger.debug("Request building") - attrs_map = { - "content": PropertyMeta( - name="content", - cast=self._content_builder.build_list, - ), - "description": PropertyMeta(name="description", cast=str), - "required": PropertyMeta(name="required", cast=bool), - } + attrs_map = { + "content": PropertyMeta( + name="content", + cast=self._content_builder.build_list, + ), + "description": PropertyMeta(name="description", cast=str), + "required": PropertyMeta(name="required", cast=bool), + } - attrs = extract_typed_props(data, attrs_map) + attrs = extract_typed_props(data, attrs_map) - return RequestBody(**attrs) + return RequestBody(**attrs) diff --git a/src/openapi_parser/builders/response.py b/src/openapi_parser/builders/response.py index 12dd6fc..7b3013f 100644 --- a/src/openapi_parser/builders/response.py +++ b/src/openapi_parser/builders/response.py @@ -3,10 +3,14 @@ import logging from typing import Any -from openapi_parser.builders.common import PropertyMeta, extract_typed_props +from openapi_parser.builders.common import ( + PropertyMeta, + extract_typed_props, +) from openapi_parser.builders.content import ContentBuilder from openapi_parser.builders.header import HeaderBuilder from openapi_parser.builders.link import LinkBuilder +from openapi_parser.logging import log_ctx from openapi_parser.specification import Response logger = logging.getLogger(__name__) @@ -36,33 +40,38 @@ def __init__( self._header_builder = header_builder self._link_builder = link_builder - def build(self, code: int | str, data: dict[str, Any]) -> Response: + def build( + self, + code: int | str, + data: dict[str, Any], + ) -> Response: """Build a Response from a status code and raw data dict.""" - logger.debug(f"Response building [code={code}]") + with log_ctx("responses", str(code)): + logger.debug(f"Response building [code={code}]") - attrs_map = { - "description": PropertyMeta(name="description", cast=str), - "content": PropertyMeta( - name="content", - cast=self._content_builder.build_list, - ), - "headers": PropertyMeta( - name="headers", - cast=self._header_builder.build_list, - ), - "links": PropertyMeta( - name="links", - cast=self._link_builder.build_dict, - ), - } + attrs_map = { + "description": PropertyMeta(name="description", cast=str), + "content": PropertyMeta( + name="content", + cast=self._content_builder.build_list, + ), + "headers": PropertyMeta( + name="headers", + cast=self._header_builder.build_list, + ), + "links": PropertyMeta( + name="links", + cast=self._link_builder.build_dict, + ), + } - attrs = extract_typed_props(data, attrs_map) + attrs = extract_typed_props(data, attrs_map) - attrs["is_default"] = code == "default" + attrs["is_default"] = code == "default" - try: - attrs["code"] = int(code) - except ValueError: - logger.debug(f"Response code is not an integer [code={code}]") + try: + attrs["code"] = int(code) + except ValueError: + logger.debug(f"Response code is not an integer [code={code}]") - return Response(**attrs) + return Response(**attrs) diff --git a/src/openapi_parser/builders/schema.py b/src/openapi_parser/builders/schema.py index 9f131bf..b890d91 100644 --- a/src/openapi_parser/builders/schema.py +++ b/src/openapi_parser/builders/schema.py @@ -37,7 +37,7 @@ String, ) -SchemaBuilderMethod = Callable[[dict[str, Any]], Schema] +SchemaBuilderMethod = Callable[..., Schema] ALL_OF_SCHEMAS_KEY = "allOf" @@ -55,7 +55,7 @@ def extract_attrs( attrs_map (Dict[str, PropertyMeta]): Type-casting mapping Returns: - Dict[str, Any]: Extracted dictionary with typed values + dict: Extracted attributes dictionary """ base_attrs_map = { "type": "type", @@ -88,7 +88,9 @@ def extract_attrs( return attrs -def merge_all_of_schemas(original_data: dict[str, Any]) -> dict[str, Any]: +def merge_all_of_schemas( + original_data: dict[str, Any], +) -> dict[str, Any]: """Recursive merge schemas with 'allOf' type into single schema dictionary. Args: @@ -237,11 +239,14 @@ def _boolean(data: dict[str, Any]) -> Boolean: return Boolean(**extract_attrs(data, {})) def _array(self, data: dict[str, Any]) -> Array: + def build_items(items_data: dict[str, Any]) -> Schema: + return self.create(items_data) + attrs_map = { "max_items": PropertyMeta(name="maxItems", cast=int), "min_items": PropertyMeta(name="minItems", cast=int), "unique_items": PropertyMeta(name="uniqueItems", cast=bool), - "items": PropertyMeta(name="items", cast=self.create), + "items": PropertyMeta(name="items", cast=build_items), } return Array(**extract_attrs(data, attrs_map)) diff --git a/src/openapi_parser/builders/schemas.py b/src/openapi_parser/builders/schemas.py index fd08268..5cb97d3 100644 --- a/src/openapi_parser/builders/schemas.py +++ b/src/openapi_parser/builders/schemas.py @@ -4,6 +4,7 @@ from typing import Any from openapi_parser.builders.schema import SchemaFactory +from openapi_parser.logging import log_ctx from openapi_parser.specification import Schema logger = logging.getLogger(__name__) @@ -22,10 +23,17 @@ def __init__(self, schema_factory: SchemaFactory) -> None: """ self._schema_factory = schema_factory - def build_collection(self, schemas: dict[str, Any]) -> dict[str, Schema]: + def build_collection( + self, + schemas: dict[str, Any], + ) -> dict[str, Schema]: """Build a dict of named Schema objects.""" logger.debug(f"Schemas parsing: {schemas.keys()}") - return { - key: self._schema_factory.create(value) for key, value in schemas.items() - } + result: dict[str, Schema] = {} + + for key, value in schemas.items(): + with log_ctx(key): + result[key] = self._schema_factory.create(value) + + return result diff --git a/src/openapi_parser/builders/security.py b/src/openapi_parser/builders/security.py index ee925de..f6f3d6f 100644 --- a/src/openapi_parser/builders/security.py +++ b/src/openapi_parser/builders/security.py @@ -10,6 +10,7 @@ ) from openapi_parser.builders.oauth_flow import OAuthFlowBuilder from openapi_parser.enumeration import AuthenticationScheme, BaseLocation, SecurityType +from openapi_parser.logging import log_ctx from openapi_parser.specification import Security logger = logging.getLogger(__name__) @@ -30,7 +31,7 @@ def __init__(self, oauth_flow_builder: OAuthFlowBuilder) -> None: def build(self, data: dict[str, Any]) -> Security: """Build a Security object from a raw dict.""" - logger.debug(f"Security item parsing [{data}]") + logger.debug("Security item parsing") attrs_map = { "type": PropertyMeta(name="type", cast=SecurityType), @@ -50,13 +51,21 @@ def build(self, data: dict[str, Any]) -> Security: attrs["extensions"] = extract_extension_attributes(data) if attrs["extensions"]: - logger.debug(f"Extracted custom properties [{attrs['extensions'].keys()}]") + logger.debug( + f"Extracted custom properties [{attrs['extensions'].keys()}]", + ) return Security(**attrs) - def build_collection(self, data: dict[str, Any]) -> dict[str, Security]: + def build_collection( + self, + data: dict[str, Any], + ) -> dict[str, Security]: """Build a dict of named Security objects.""" - return { - scheme_name: self.build(scheme_data) - for scheme_name, scheme_data in data.items() - } + result: dict[str, Security] = {} + + for scheme_name, scheme_data in data.items(): + with log_ctx("securitySchemes", scheme_name): + result[scheme_name] = self.build(scheme_data) + + return result diff --git a/src/openapi_parser/builders/server.py b/src/openapi_parser/builders/server.py index d201b34..50454f0 100644 --- a/src/openapi_parser/builders/server.py +++ b/src/openapi_parser/builders/server.py @@ -9,6 +9,7 @@ extract_typed_props, ) from openapi_parser.errors import ParserError +from openapi_parser.logging import log_ctx from openapi_parser.specification import Server logger = logging.getLogger(__name__) @@ -17,7 +18,10 @@ class ServerBuilder: """Builds server objects from raw specification data.""" - def build_list(self, data_list: list[dict[str, Any]]) -> list[Server]: + def build_list( + self, + data_list: list[dict[str, Any]], + ) -> list[Server]: """Build a list of Server objects from a list of raw dicts.""" return [self._build_server(item) for item in data_list] @@ -25,21 +29,26 @@ def build_list(self, data_list: list[dict[str, Any]]) -> list[Server]: def _build_server(data: dict[str, Any]) -> Server: url = data.get("url") - if url is None: - raise ParserError("Server definition is missing required 'url' property") + with log_ctx("servers", url): + if url is None: + raise ParserError( + "Server definition is missing required 'url' property" + ) - logger.debug(f"Server item parsing [{url}]") + logger.debug(f"Server item parsing [{url}]") - attrs_map = { - "url": PropertyMeta(name="url", cast=str), - "description": PropertyMeta(name="description", cast=str), - "variables": PropertyMeta(name="variables", cast=dict), - } + attrs_map = { + "url": PropertyMeta(name="url", cast=str), + "description": PropertyMeta(name="description", cast=str), + "variables": PropertyMeta(name="variables", cast=dict), + } - attrs = extract_typed_props(data, attrs_map) - attrs["extensions"] = extract_extension_attributes(data) + attrs = extract_typed_props(data, attrs_map) + attrs["extensions"] = extract_extension_attributes(data) - if attrs["extensions"]: - logger.debug(f"Extracted custom properties [{attrs['extensions'].keys()}]") + if attrs["extensions"]: + logger.debug( + f"Extracted custom properties [{attrs['extensions'].keys()}]" + ) - return Server(**attrs) + return Server(**attrs) diff --git a/src/openapi_parser/builders/tag.py b/src/openapi_parser/builders/tag.py index 40262b6..6eda834 100644 --- a/src/openapi_parser/builders/tag.py +++ b/src/openapi_parser/builders/tag.py @@ -3,9 +3,13 @@ import logging from typing import Any -from openapi_parser.builders.common import PropertyMeta, extract_typed_props +from openapi_parser.builders.common import ( + PropertyMeta, + extract_typed_props, +) from openapi_parser.builders.external_doc import ExternalDocBuilder from openapi_parser.errors import ParserError +from openapi_parser.logging import log_ctx from openapi_parser.specification import Tag logger = logging.getLogger(__name__) @@ -24,27 +28,32 @@ def __init__(self, external_doc_builder: ExternalDocBuilder) -> None: """ self._external_doc_builder = external_doc_builder - def build_list(self, data_list: list[dict[str, Any]]) -> list[Tag]: + def build_list( + self, + data_list: list[dict[str, Any]], + ) -> list[Tag]: """Build a list of Tag objects from a list of raw dicts.""" return [self._build_tag(item) for item in data_list] def _build_tag(self, data: dict[str, Any]) -> Tag: - name = data.get("name") + with log_ctx("tags"): + name = data.get("name") - if name is None: - raise ParserError("Tag is missing required 'name' property") + if name is None: + raise ParserError("Tag is missing required 'name' property") - logger.debug(f"Tag building [{name}]") + with log_ctx(name): + logger.debug(f"Tag building [{name}]") - attrs_map = { - "name": PropertyMeta(name="name", cast=str), - "description": PropertyMeta(name="description", cast=str), - "external_docs": PropertyMeta( - name="externalDocs", - cast=self._external_doc_builder.build, - ), - } + attrs_map = { + "name": PropertyMeta(name="name", cast=str), + "description": PropertyMeta(name="description", cast=str), + "external_docs": PropertyMeta( + name="externalDocs", + cast=self._external_doc_builder.build, + ), + } - attrs = extract_typed_props(data, attrs_map) + attrs = extract_typed_props(data, attrs_map) - return Tag(**attrs) + return Tag(**attrs) diff --git a/src/openapi_parser/errors.py b/src/openapi_parser/errors.py index 126a559..66cfadc 100644 --- a/src/openapi_parser/errors.py +++ b/src/openapi_parser/errors.py @@ -1,5 +1,7 @@ """Custom exceptions for OpenAPI parsing errors.""" +from openapi_parser.logging import _log_ctx_var + class ParserError(Exception): """Base parser exception class. @@ -7,4 +9,23 @@ class ParserError(Exception): Throws when any error occurs. """ - pass + context: str | None + + def __init__(self, message: str, context: str | None = None) -> None: + """Initialize the error with an optional parse context path. + + Args: + message: Error description + context: Path within the spec where the error occurred (e.g. "paths./users.get") + """ + super().__init__(message) + self.context = context if context is not None else (_log_ctx_var.get()) + + def __str__(self) -> str: + """Format error message with optional context prefix.""" + msg = super().__str__() + + if self.context: + return f"[{self.context}] {msg}" + + return msg diff --git a/src/openapi_parser/logging.py b/src/openapi_parser/logging.py new file mode 100644 index 0000000..d9b4539 --- /dev/null +++ b/src/openapi_parser/logging.py @@ -0,0 +1,79 @@ +"""Logging utilities for automatic context prefixing.""" + +from __future__ import annotations + +import contextvars +import logging + +_log_ctx_var: contextvars.ContextVar[str] = contextvars.ContextVar( + "openapi_log_ctx", + default="", +) + +_original_factory = logging.getLogRecordFactory() + + +def _context_record_factory( + name: str, + level: int, + pathname: str, + lineno: int, + msg: str, + args: tuple[object, ...], + exc_info: object, + func: str | None = None, + sinfo: str | None = None, + **kwargs: object, +) -> logging.LogRecord: + record = _original_factory( + name, + level, + pathname, + lineno, + msg, + args, + exc_info, + func, + sinfo, + **kwargs, + ) + + ctx = _log_ctx_var.get() + + if ctx: + record.msg = f"[{ctx}] {record.msg}" + + return record + + +logging.setLogRecordFactory(_context_record_factory) + + +class log_ctx: + """Context manager that appends segments to the current parse context. + + Usage: + with log_ctx("paths", url): + ... + with log_ctx("get"): + ... + """ + + def __init__(self, *segments: str | None) -> None: + """Initialize context manager with path segments to append.""" + self.segments = [s for s in segments if s is not None] + + def __enter__(self) -> log_ctx: + """Enter context block, appending segments to the current context.""" + current = _log_ctx_var.get() + + parts = [current] if current else [] + parts.extend(self.segments) + + self._token = _log_ctx_var.set(".".join(parts)) + + return self + + def __exit__(self, *_args: object) -> None: + """Exit context block, restoring the previous context.""" + _log_ctx_var.reset(self._token) diff --git a/src/openapi_parser/parser.py b/src/openapi_parser/parser.py index a0a7476..0c82df3 100644 --- a/src/openapi_parser/parser.py +++ b/src/openapi_parser/parser.py @@ -22,6 +22,7 @@ from openapi_parser.builders.server import ServerBuilder from openapi_parser.builders.tag import TagBuilder from openapi_parser.errors import ParserError +from openapi_parser.logging import log_ctx from openapi_parser.resolver import OpenAPIResolver from openapi_parser.specification import Specification @@ -80,61 +81,64 @@ def load_specification(self, data: dict[str, Any]) -> Specification: Raises: ParserError: If OpenAPI schema is invalid """ - logger.debug("Building Specification objects") - - try: - version = data["openapi"] - except KeyError: - raise ParserError( - "Invalid OpenAPI version, check 'openapi' property in the document", - ) from None - - attrs_map = { - "servers": PropertyMeta( - name="servers", - cast=self.server_builder.build_list, - ), - "tags": PropertyMeta( - name="tags", - cast=self.tag_builder.build_list, - ), - "external_docs": PropertyMeta( - name="externalDocs", - cast=self.external_doc_builder.build, - ), - "paths": PropertyMeta( - name="paths", - cast=self.path_builder.build_list, - ), - "security": PropertyMeta(name="security", cast=None), - } - - attrs = extract_typed_props(data, attrs_map) - - attrs["version"] = version - - info_data = data.get("info") - - if info_data is None: - raise ParserError("OpenAPI document is missing required 'info' property") - - attrs["info"] = self.info_builder.build(info_data) - - components = data.get("components") or {} - - if "securitySchemes" in components: - attrs["security_schemas"] = self.security_builder.build_collection( - components["securitySchemes"], - ) - - if "schemas" in components: - attrs["schemas"] = self.schemas_builder.build_collection( - components["schemas"], - ) - - logger.debug("Specification parsed successfully") - - return Specification(**attrs) + with log_ctx("spec"): + logger.debug("Building Specification objects") + + try: + version = data["openapi"] + except KeyError: + raise ParserError( + "Invalid OpenAPI version, check 'openapi' property in the document", + ) from None + + attrs_map = { + "servers": PropertyMeta( + name="servers", + cast=self.server_builder.build_list, + ), + "tags": PropertyMeta( + name="tags", + cast=self.tag_builder.build_list, + ), + "external_docs": PropertyMeta( + name="externalDocs", + cast=self.external_doc_builder.build, + ), + "paths": PropertyMeta( + name="paths", + cast=self.path_builder.build_list, + ), + "security": PropertyMeta(name="security", cast=None), + } + + attrs = extract_typed_props(data, attrs_map) + + attrs["version"] = version + + info_data = data.get("info") + + if info_data is None: + raise ParserError( + "OpenAPI document is missing required 'info' property" + ) + + attrs["info"] = self.info_builder.build(info_data) + + components = data.get("components") or {} + + if "securitySchemes" in components: + attrs["security_schemas"] = self.security_builder.build_collection( + components["securitySchemes"], + ) + + if "schemas" in components: + attrs["schemas"] = self.schemas_builder.build_collection( + components["schemas"], + ) + + logger.debug("Specification parsed successfully") + + return Specification(**attrs) def _create_parser(strict_enum: bool = True) -> Parser: diff --git a/tests/builders/test_content_builder.py b/tests/builders/test_content_builder.py index 9b82a8e..277767c 100644 --- a/tests/builders/test_content_builder.py +++ b/tests/builders/test_content_builder.py @@ -168,7 +168,7 @@ def test_build_with_encoding() -> None: "name": Encoding(content_type="text/plain"), } encoding_builder.build_dict.assert_called_once_with( - {"name": {"contentType": "text/plain"}} + {"name": {"contentType": "text/plain"}}, ) diff --git a/tests/test_logging.py b/tests/test_logging.py new file mode 100644 index 0000000..c299f34 --- /dev/null +++ b/tests/test_logging.py @@ -0,0 +1,27 @@ +import logging + +from openapi_parser.logging import log_ctx + + +def test_log_context_prefix() -> None: + logger = logging.getLogger(__name__) + handler = logging.StreamHandler() + handler.setFormatter(logging.Formatter("%(message)s")) + records: list[logging.LogRecord] = [] + + class RecordHandler(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + record.msg = self.format(record) + records.append(record) + + record_handler = RecordHandler() + logger.addHandler(record_handler) + logger.setLevel(logging.DEBUG) + + with log_ctx("test", "path"): + logger.debug("hello") + + logger.removeHandler(record_handler) + + assert len(records) == 1 + assert records[0].msg == "[test.path] hello"