🔀 Merge branch 'feature/response-envelope' into develop

Author: rafalkrupinski

.gitignore

@@ -1 +1,2 @@
 /dist/
+__pycache__/

src/lapidary/render/main.py

@@ -137,7 +137,7 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> bool | None:
 
     def writing_file(self, template: pathlib.PurePath, target: pathlib.Path) -> None:
         super().writing_file(template, target)
-        if str(template) == '{{model.package}}/{{loop_over(model.modules).rel_path}}.py.jinja':
+        if str(template) == '{{loop_over(model.modules).file_path}}.py.jinja':
             self._progress_bar.update(1, str(target).split('/')[-2])
 
 

src/lapidary/render/model/openapi/model.py

@@ -329,7 +329,7 @@ def _validate(values: Mapping[str, typing.Any]):
 
 class Response(ExtendableModel):
     description: str
-    headers: dict[str, Reference[Header] | Header] | None = None
+    headers: Annotated[dict[str, Reference[Header] | Header], pydantic.Field(default_factory=dict)]
     content: 'typing.Annotated[dict[str, MediaType], pydantic.Field(default_factory=dict)]'
     links: dict[str, Reference[Link] | Link] | None = None
 

src/lapidary/render/model/openapi_conv.py

@@ -1,14 +1,14 @@
 import functools
 import logging
-from collections.abc import Callable, Iterable, Mapping
+from collections.abc import Callable, Iterable, Mapping, MutableMapping
 from typing import Any, cast
 
 from mimeparse import parse_media_range
 
 from .. import json_pointer, names
 from . import openapi, python
 from .refs import resolve_ref
-from .schema import OpenApi30SchemaConverter
+from .schema import OpenApi30SchemaConverter, resolve_type_hint
 from .stack import Stack
 
 logger = logging.getLogger(__name__)
@@ -39,6 +39,8 @@ def __init__(
             package=str(root_package),
         )
 
+        self._response_mime_maps: MutableMapping[Stack, python.MimeMap] = {}
+
     def process(self) -> python.ClientModel:
         stack = Stack()
 
@@ -98,25 +100,35 @@ def process_global_responses(self, value: openapi.Responses | None, stack: Stack
     # Not providing process_parameters (plural) as each caller calls it in a different context
     # (list, map or map with defaults)
 
-    @resolve_ref
-    def process_parameter(self, value: openapi.Parameter, stack: Stack) -> python.Parameter:
-        logger.debug('process_parameter %s', stack)
-
-        if not isinstance(value, openapi.ParameterBase):
-            raise TypeError(f'Expected Parameter object at {stack}, got {type(value).__name__}.')
+    def _process_schema_or_content(
+        self,
+        value: openapi.ParameterBase,
+        stack: Stack,
+    ) -> tuple[python.TypeHint, str]:
+        if value.schema_ and value.content:
+            raise ValueError()
         if value.schema_:
             media_type: str | None = None
             # encoding: str | None = None
-            typ = self.schema_converter.process_schema(value.schema_, stack.push('schema'), value.required)
+            return self.schema_converter.process_schema(value.schema_, stack.push('schema'), value.required), media_type
         elif value.content:
             media_type, media_type_obj = next(iter(value.content.items()))
             # encoding = media_type_obj.encoding
-            typ = self.schema_converter.process_schema(
+            return self.schema_converter.process_schema(
                 media_type_obj.schema_, stack.push('content', media_type), value.required
-            )
+            ), media_type
         else:
             raise TypeError(f'{stack}: schema or content is required')
 
+    @resolve_ref
+    def process_parameter(self, value: openapi.Parameter, stack: Stack) -> python.Parameter:
+        logger.debug('process_parameter %s', stack)
+
+        if not isinstance(value, openapi.ParameterBase):
+            raise TypeError(f'Expected Parameter object at {stack}, got {type(value).__name__}.')
+
+        typ, media_type = self._process_schema_or_content(value, stack)
+
         return python.Parameter(
             name=parameter_name(value),
             alias=value.name,
@@ -161,9 +173,56 @@ def process_response(
         stack: Stack,
     ) -> python.MimeMap:
         assert isinstance(value, openapi.Response)
-        return self.process_content(value.content, stack.push('content'))
+
+        if stack in self._response_mime_maps:
+            return self._response_mime_maps[stack]
+
+        headers_stack = stack.push('headers')
+
+        headers = [self.process_header(header, headers_stack.push(name)) for name, header in value.headers.items()]
+
+        mime_map_body_only = self.process_content(value.content, stack.push('content'))
+        mime_map = {
+            mime_type: self._create_envelope_model(
+                body_type=body_type,
+                headers=headers,
+                stack=stack,
+            )
+            for mime_type, body_type in mime_map_body_only.items()
+        }
+        self._response_mime_maps[stack] = mime_map
+        return mime_map
+
+    @resolve_ref
+    def process_header(self, value: openapi.Header, stack: Stack) -> python.ResponseHeader:
+        alias = stack.top()
+
+        typ, _ = self._process_schema_or_content(value, stack)
+
+        return python.ResponseHeader(name=names.maybe_mangle_name(alias), alias=alias, type=typ, annotation='Header')
+
+    def _create_envelope_model(
+        self,
+        body_type: python.TypeHint,
+        headers: Iterable[python.ResponseHeader],
+        stack: Stack,
+    ) -> python.TypeHint:
+        envelope = python.ResponseEnvelopeModel(
+            name='Response',
+            headers=headers,
+            body_type=body_type,
+        )
+        type_hint = resolve_type_hint(str(self.root_package), stack.push('Response'))
+        self.target.add_response_envelope_module(
+            python.ResponseEnvelopeModule(
+                path=python.ModulePath(type_hint.module, is_module=False),
+                body=envelope,
+            )
+        )
+        return type_hint
 
     def process_content(self, value: Mapping[str, openapi.MediaType], stack: Stack) -> python.MimeMap:
+        """Returns: {mime_type: response body type hint}"""
         types = {}
         for mime, media_type in value.items():
             mime_parsed = parse_media_range(mime)

src/lapidary/render/model/python/__init__.py

@@ -22,6 +22,8 @@
     'ParamLocation',
     'Parameter',
     'PasswordOAuth2Flow',
+    'ResponseEnvelopeModel',
+    'ResponseHeader',
     'ResponseMap',
     'SchemaClass',
     'SchemaModule',
@@ -31,6 +33,7 @@
 )
 
 import dataclasses as dc
+import itertools
 from collections.abc import Iterable, MutableMapping, MutableSequence, MutableSet, Sequence
 from functools import cached_property
 from typing import Self
@@ -53,11 +56,13 @@
     OperationFunction,
     Parameter,
     PasswordOAuth2Flow,
+    ResponseEnvelopeModel,
+    ResponseHeader,
     ResponseMap,
     SchemaClass,
     SecurityRequirements,
 )
-from .module import AbstractModule, AuthModule, ClientModule, EmptyModule, SchemaModule
+from .module import AbstractModule, AuthModule, ClientModule, EmptyModule, ResponseEnvelopeModule, SchemaModule
 from .module_path import ModulePath
 from .type_hint import NONE, BuiltinTypeHint, GenericTypeHint, TypeHint, type_hint_or_union
 
@@ -68,18 +73,18 @@ class ClientModel:
     package: str
     schemas: MutableSequence[SchemaModule] = dc.field(default_factory=list)
     security_schemes: MutableMapping[str, Auth] = dc.field(default_factory=dict)
+    _response_envelopes: MutableSequence[ResponseEnvelopeModule] = dc.field(default_factory=list)
 
     def packages(self: Self) -> Iterable[ModulePath]:
         # Used to create __init__.py files in otherwise empty packages
 
-        known_packages: MutableSet[ModulePath] = {ModulePath.root()}
+        known_packages: MutableSet[ModulePath] = {ModulePath(self.package)}
 
-        # for each schema module get its package
-        for schema in self.schemas:
-            path: ModulePath | None = schema.path
+        for mod in itertools.chain(self.schemas, self._response_envelopes):
+            path: ModulePath | None = mod.path
             while path := path.parent():
                 if path in known_packages:
-                    continue
+                    break
                 yield path
                 known_packages.add(path)
 
@@ -88,6 +93,15 @@ def modules(self) -> Sequence[AbstractModule]:
         return list(self._modules())
 
     def _modules(self) -> Iterable[AbstractModule]:
-        yield from self.schemas
+        known_modules = set()
+        for mod in itertools.chain(self.schemas, self._response_envelopes):
+            assert mod.path not in known_modules, mod.path
+            known_modules.add(mod.path)
+            yield mod
+
         for package in self.packages():
-            yield EmptyModule(path=package, body=None)
+            if package not in known_modules:
+                yield EmptyModule(path=package, body=None)
+
+    def add_response_envelope_module(self, mod: ResponseEnvelopeModule):
+        self._response_envelopes.append(mod)

src/lapidary/render/model/python/model.py

@@ -1,7 +1,7 @@
 import dataclasses as dc
 import enum
 from collections.abc import Iterable, Mapping
-from typing import Any, TypeAlias
+from typing import Any, Literal, TypeAlias
 
 from ..openapi import ParameterLocation as ParamLocation
 from ..openapi import Style as ParamStyle
@@ -35,10 +35,6 @@ class Field:
     deprecated: bool = False
     """Currently not used"""
 
-    @property
-    def dependencies(self) -> Iterable[TypeHint]:
-        return [self.annotation.type]
-
 
 @dc.dataclass
 class Auth:
@@ -119,11 +115,10 @@ class OperationFunction:
     responses: ResponseMap
     security: SecurityRequirements | None
 
-    @property
     def dependencies(self) -> Iterable[TypeHint]:
         yield self.request_body_type
         for param in self.params:
-            yield from param.dependencies
+            yield from param.dependencies()
         yield self.response_body_type
 
     @property
@@ -164,7 +159,6 @@ class Parameter:
     style: ParamStyle | None
     explode: bool | None
 
-    @property
     def dependencies(self) -> Iterable[TypeHint]:
         yield self.type
 
@@ -184,11 +178,10 @@ class SchemaClass:
     fields: list[Field] = dc.field(default_factory=list)
     model_type: ModelType = ModelType.model
 
-    @property
     def dependencies(self) -> Iterable[TypeHint]:
         yield self.base_type
         for prop in self.fields:
-            yield from prop.dependencies
+            yield prop.annotation.type
 
 
 @dc.dataclass
@@ -200,7 +193,6 @@ class ClientInit:
     response_map: ResponseMap = dc.field(default_factory=dict)
     security: SecurityRequirements | None = None
 
-    @property
     def dependencies(self) -> Iterable[TypeHint]:
         for mime_map in self.response_map.values():
             yield from mime_map.values()
@@ -211,8 +203,30 @@ class ClientClass:
     init_method: ClientInit
     methods: list[OperationFunction] = dc.field(default_factory=list)
 
-    @property
     def dependencies(self) -> Iterable[TypeHint]:
-        yield from self.init_method.dependencies
+        yield from self.init_method.dependencies()
         for fn in self.methods:
-            yield from fn.dependencies
+            yield from fn.dependencies()
+
+
+@dc.dataclass(frozen=True)
+class ResponseHeader:
+    name: str
+    alias: str
+    type: TypeHint
+    annotation: Literal['Cookie', 'Header', 'Link']
+
+    def dependencies(self) -> Iterable[TypeHint]:
+        yield self.type
+
+
+@dc.dataclass(frozen=True)
+class ResponseEnvelopeModel:
+    name: str
+    headers: Iterable[ResponseHeader]
+    body_type: TypeHint
+
+    def dependencies(self) -> Iterable[TypeHint]:
+        yield self.body_type
+        for header in self.headers:
+            yield from header.dependencies()