openapi-core/openapi_core/unmarshalling/unmarshallers.py at 96d51c6f90001856433a7f482da74380d965b4a6 · python-openapi/openapi-core · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
from typing import Any
from typing import Mapping
from typing import Optional
from typing import Tuple

from jsonschema_path import SchemaPath
from openapi_spec_validator.validation.types import SpecValidatorType

from openapi_core.casting.schemas.factories import SchemaCastersFactory
from openapi_core.deserializing.media_types.datatypes import (
    MediaTypeDeserializersDict,
)
from openapi_core.deserializing.media_types.factories import (
    MediaTypeDeserializersFactory,
)
from openapi_core.deserializing.styles.factories import (
    StyleDeserializersFactory,
)
from openapi_core.templating.paths.types import PathFinderType
from openapi_core.unmarshalling.schemas.datatypes import (
    FormatUnmarshallersDict,
)
from openapi_core.unmarshalling.schemas.factories import (
    SchemaUnmarshallersFactory,
)
from openapi_core.validation.schemas.datatypes import FormatValidatorsDict
from openapi_core.validation.schemas.factories import SchemaValidatorsFactory
from openapi_core.validation.validators import BaseValidator


class BaseUnmarshaller(BaseValidator):
    schema_unmarshallers_factory: SchemaUnmarshallersFactory = NotImplemented

    def __init__(
        self,
        spec: SchemaPath,
        base_url: Optional[str] = None,
        style_deserializers_factory: Optional[
            StyleDeserializersFactory
        ] = None,
        media_type_deserializers_factory: Optional[
            MediaTypeDeserializersFactory
        ] = None,
        schema_casters_factory: Optional[SchemaCastersFactory] = None,
        schema_validators_factory: Optional[SchemaValidatorsFactory] = None,
        path_finder_cls: Optional[PathFinderType] = None,
        spec_validator_cls: Optional[SpecValidatorType] = None,
        format_validators: Optional[FormatValidatorsDict] = None,
        extra_format_validators: Optional[FormatValidatorsDict] = None,
        extra_media_type_deserializers: Optional[
            MediaTypeDeserializersDict
        ] = None,
        strict_additional_properties: bool = False,
        enforce_properties_required: bool = False,
        schema_unmarshallers_factory: Optional[
            SchemaUnmarshallersFactory
        ] = None,
        format_unmarshallers: Optional[FormatUnmarshallersDict] = None,
        extra_format_unmarshallers: Optional[FormatUnmarshallersDict] = None,
    ):
        if schema_validators_factory is None and schema_unmarshallers_factory:
            schema_validators_factory = (
                schema_unmarshallers_factory.schema_validators_factory
            )
        BaseValidator.__init__(
            self,
            spec,
            base_url=base_url,
            style_deserializers_factory=style_deserializers_factory,
            media_type_deserializers_factory=media_type_deserializers_factory,
            schema_casters_factory=schema_casters_factory,
            schema_validators_factory=schema_validators_factory,
            path_finder_cls=path_finder_cls,
            spec_validator_cls=spec_validator_cls,
            format_validators=format_validators,
            extra_format_validators=extra_format_validators,
            extra_media_type_deserializers=extra_media_type_deserializers,
            strict_additional_properties=strict_additional_properties,
            enforce_properties_required=enforce_properties_required,
        )
        self.schema_unmarshallers_factory = (
            schema_unmarshallers_factory or self.schema_unmarshallers_factory
        )
        if self.schema_unmarshallers_factory is NotImplemented:
            raise NotImplementedError(
                "schema_unmarshallers_factory is not assigned"
            )
        self.format_unmarshallers = format_unmarshallers
        self.extra_format_unmarshallers = extra_format_unmarshallers

    def _unmarshal_schema(self, schema: SchemaPath, value: Any) -> Any:
        unmarshaller = self.schema_unmarshallers_factory.create(
            schema,
            format_validators=self.format_validators,
            extra_format_validators=self.extra_format_validators,
            strict_additional_properties=self.strict_additional_properties,
            enforce_properties_required=self.enforce_properties_required,
            format_unmarshallers=self.format_unmarshallers,
            extra_format_unmarshallers=self.extra_format_unmarshallers,
        )
        return unmarshaller.unmarshal(value)

    def _get_param_or_header_and_schema(
        self,
        param_or_header: SchemaPath,
        location: Mapping[str, Any],
        name: Optional[str] = None,
    ) -> Tuple[Any, Optional[SchemaPath]]:
        casted, schema = super()._get_param_or_header_and_schema(
            param_or_header, location, name=name
        )
        if schema is None:
            return casted, None
        return self._unmarshal_schema(schema, casted), schema

    def _get_content_and_schema(
        self, raw: Any, content: SchemaPath, mimetype: Optional[str] = None
    ) -> Tuple[Any, Optional[SchemaPath]]:
        casted, schema = super()._get_content_and_schema(
            raw, content, mimetype
        )
        if schema is None:
            return casted, None
        return self._unmarshal_schema(schema, casted), schema