1from .common import resolve_schema_instance 2 3 4class SchemaResolver: 5 """Resolve marshmallow Schemas in OpenAPI components and translate to OpenAPI 6 `schema objects 7 <https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#schema-object>`_, 8 `parameter objects 9 <https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#parameter-object>`_ 10 or `reference objects 11 <https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#reference-object>`_. 12 """ 13 14 def __init__(self, openapi_version, converter): 15 self.openapi_version = openapi_version 16 self.converter = converter 17 18 def resolve_operations(self, operations, **kwargs): 19 """Resolve marshmallow Schemas in a dict mapping operation to OpenApi `Operation Object 20 https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#operationObject`_""" 21 22 for operation in operations.values(): 23 if not isinstance(operation, dict): 24 continue 25 if "parameters" in operation: 26 operation["parameters"] = self.resolve_parameters( 27 operation["parameters"] 28 ) 29 if self.openapi_version.major >= 3: 30 self.resolve_callback(operation.get("callbacks", {})) 31 if "requestBody" in operation: 32 self.resolve_schema(operation["requestBody"]) 33 for response in operation.get("responses", {}).values(): 34 self.resolve_response(response) 35 36 def resolve_callback(self, callbacks): 37 """Resolve marshmallow Schemas in a dict mapping callback name to OpenApi `Callback Object 38 https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#callbackObject`_. 39 40 This is done recursively, so it is possible to define callbacks in your callbacks. 41 42 Example: :: 43 44 #Input 45 { 46 "userEvent": { 47 "https://my.example/user-callback": { 48 "post": { 49 "requestBody": { 50 "content": { 51 "application/json": { 52 "schema": UserSchema 53 } 54 } 55 } 56 }, 57 } 58 } 59 } 60 61 #Output 62 { 63 "userEvent": { 64 "https://my.example/user-callback": { 65 "post": { 66 "requestBody": { 67 "content": { 68 "application/json": { 69 "schema": { 70 "$ref": "#/components/schemas/User" 71 } 72 } 73 } 74 } 75 }, 76 } 77 } 78 } 79 80 81 """ 82 for callback in callbacks.values(): 83 if isinstance(callback, dict): 84 for path in callback.values(): 85 self.resolve_operations(path) 86 87 def resolve_parameters(self, parameters): 88 """Resolve marshmallow Schemas in a list of OpenAPI `Parameter Objects 89 <https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#parameter-object>`_. 90 Each parameter object that contains a Schema will be translated into 91 one or more Parameter Objects. 92 93 If the value of a `schema` key is marshmallow Schema class, instance or 94 a string that resolves to a Schema Class each field in the Schema will 95 be expanded as a separate Parameter Object. 96 97 Example: :: 98 99 #Input 100 class UserSchema(Schema): 101 name = fields.String() 102 id = fields.Int() 103 104 [ 105 {"in": "query", "schema": "UserSchema"} 106 ] 107 108 #Output 109 [ 110 {"in": "query", "name": "id", "required": False, "schema": {"type": "integer"}}, 111 {"in": "query", "name": "name", "required": False, "schema": {"type": "string"}} 112 ] 113 114 If the Parameter Object contains a `content` key a single Parameter 115 Object is returned with the Schema translated into a Schema Object or 116 Reference Object. 117 118 Example: :: 119 120 #Input 121 [{"in": "query", "name": "pet", "content":{"application/json": {"schema": "PetSchema"}} }] 122 123 #Output 124 [ 125 { 126 "in": "query", 127 "name": "pet", 128 "content": { 129 "application/json": { 130 "schema": {"$ref": "#/components/schemas/Pet"} 131 } 132 } 133 } 134 ] 135 136 137 :param list parameters: the list of OpenAPI parameter objects to resolve. 138 """ 139 resolved = [] 140 for parameter in parameters: 141 if ( 142 isinstance(parameter, dict) 143 and not isinstance(parameter.get("schema", {}), dict) 144 and "in" in parameter 145 ): 146 schema_instance = resolve_schema_instance(parameter.pop("schema")) 147 resolved += self.converter.schema2parameters( 148 schema_instance, location=parameter.pop("in"), **parameter 149 ) 150 else: 151 self.resolve_schema(parameter) 152 resolved.append(parameter) 153 return resolved 154 155 def resolve_response(self, response): 156 """Resolve marshmallow Schemas in OpenAPI `Response Objects 157 <https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#responseObject>`_. 158 Schemas may appear in either a Media Type Object or a Header Object. 159 160 Example: :: 161 162 #Input 163 { 164 "content": {"application/json": {"schema": "PetSchema"}}, 165 "description": "successful operation", 166 "headers": {"PetHeader": {"schema": "PetHeaderSchema"}}, 167 } 168 169 #Output 170 { 171 "content": { 172 "application/json":{"schema": {"$ref": "#/components/schemas/Pet"}} 173 }, 174 "description": "successful operation", 175 "headers": { 176 "PetHeader": {"schema": {"$ref": "#/components/schemas/PetHeader"}} 177 }, 178 } 179 180 :param dict response: the response object to resolve. 181 """ 182 self.resolve_schema(response) 183 if "headers" in response: 184 for header in response["headers"].values(): 185 self.resolve_schema(header) 186 187 def resolve_schema(self, data): 188 """Resolve marshmallow Schemas in an OpenAPI component or header - 189 modifies the input dictionary to translate marshmallow Schemas to OpenAPI 190 Schema Objects or Reference Objects. 191 192 OpenAPIv3 Components: :: 193 194 #Input 195 { 196 "description": "user to add to the system", 197 "content": { 198 "application/json": { 199 "schema": "UserSchema" 200 } 201 } 202 } 203 204 #Output 205 { 206 "description": "user to add to the system", 207 "content": { 208 "application/json": { 209 "schema": { 210 "$ref": "#/components/schemas/User" 211 } 212 } 213 } 214 } 215 216 :param dict|str data: either a parameter or response dictionary that may 217 contain a schema, or a reference provided as string 218 """ 219 if not isinstance(data, dict): 220 return 221 222 # OAS 2 component or OAS 3 parameter or header 223 if "schema" in data: 224 data["schema"] = self.resolve_schema_dict(data["schema"]) 225 # OAS 3 component except header 226 if self.openapi_version.major >= 3: 227 if "content" in data: 228 for content in data["content"].values(): 229 if "schema" in content: 230 content["schema"] = self.resolve_schema_dict(content["schema"]) 231 232 def resolve_schema_dict(self, schema): 233 """Resolve a marshmallow Schema class, object, or a string that resolves 234 to a Schema class or a schema reference or an OpenAPI Schema Object 235 containing one of the above to an OpenAPI Schema Object or Reference Object. 236 237 If the input is a marshmallow Schema class, object or a string that resolves 238 to a Schema class the Schema will be translated to an OpenAPI Schema Object 239 or Reference Object. 240 241 Example: :: 242 243 #Input 244 "PetSchema" 245 246 #Output 247 {"$ref": "#/components/schemas/Pet"} 248 249 If the input is a dictionary representation of an OpenAPI Schema Object 250 recursively search for a marshmallow Schemas to resolve. For `"type": "array"`, 251 marshmallow Schemas may appear as the value of the `items` key. For 252 `"type": "object"` Marshmalow Schemas may appear as values in the `properties` 253 dictionary. 254 255 Examples: :: 256 257 #Input 258 {"type": "array", "items": "PetSchema"} 259 260 #Output 261 {"type": "array", "items": {"$ref": "#/components/schemas/Pet"}} 262 263 #Input 264 {"type": "object", "properties": {"pet": "PetSchcema", "user": "UserSchema"}} 265 266 #Output 267 { 268 "type": "object", 269 "properties": { 270 "pet": {"$ref": "#/components/schemas/Pet"}, 271 "user": {"$ref": "#/components/schemas/User"} 272 } 273 } 274 275 :param string|Schema|dict schema: the schema to resolve. 276 """ 277 if isinstance(schema, dict): 278 if schema.get("type") == "array" and "items" in schema: 279 schema["items"] = self.resolve_schema_dict(schema["items"]) 280 if schema.get("type") == "object" and "properties" in schema: 281 schema["properties"] = { 282 k: self.resolve_schema_dict(v) 283 for k, v in schema["properties"].items() 284 } 285 for keyword in ("oneOf", "anyOf", "allOf"): 286 if keyword in schema: 287 schema[keyword] = [ 288 self.resolve_schema_dict(s) for s in schema[keyword] 289 ] 290 if "not" in schema: 291 schema["not"] = self.resolve_schema_dict(schema["not"]) 292 return schema 293 294 return self.converter.resolve_nested_schema(schema) 295