from .resource import Resource from .utilities import rupdate from apispec import APISpec from apispec.ext.marshmallow import MarshmallowPlugin from .fields import Field from marshmallow import Schema as BaseSchema from collections import Mapping def view2path(rule: str, view: Resource, spec: APISpec): params = { "path": rule, # TODO: Validate this slightly (leading / etc) "operations": view2operations(view, spec), } if hasattr(view, "__apispec__"): # Recursively update params rupdate(params, view.__apispec__) return params def view2operations(view: Resource, spec: APISpec): ops = {} for method in Resource.methods: if hasattr(view, method): ops[method] = {} if hasattr(getattr(view, method), "__apispec__"): ops[method] = doc2operation(getattr(view, method).__apispec__, spec) return ops def doc2operation(apispec: dict, spec: APISpec): op = {} if "_params" in apispec: op["requestBody"] = { "content": { "application/json": { "schema": convert_schema(apispec.get("_params"), spec) } }, } if "_schema" in apispec: op["responses"] = { 200: { "description": "success", "content": { "application/json": { "schema": convert_schema(apispec.get("_schema"), spec) } }, } } for key, val in apispec.items(): if not key in ["_params", "_schema"]: op[key] = val return op def convert_schema(schema, spec: APISpec): if isinstance(schema, BaseSchema): return schema elif isinstance(schema, Mapping): return map2properties(schema, spec) else: raise TypeError("Unsupported schema type. Ensure schema is a Schema class, or dictionary of Field objects") def map2properties(schema, spec: APISpec): marshmallow_plugin = next( plugin for plugin in spec.plugins if isinstance(plugin, MarshmallowPlugin) ) converter = marshmallow_plugin.converter d = {} for k, v in schema.items(): if isinstance(v, Field): d[k] = converter.field2property(v) elif isinstance(v, Mapping): d[k] = map2properties(v, spec) else: d[k] = v return {"properties": d}