1import json 2 3from coreapi.codecs.base import BaseCodec 4from coreapi.compat import force_bytes 5from coreapi.document import Document 6from coreapi.exceptions import ParseError 7from openapi_codec.encode import generate_swagger_object 8from openapi_codec.decode import _parse_document 9 10 11__version__ = '1.3.2' 12 13 14class OpenAPICodec(BaseCodec): 15 media_type = 'application/openapi+json' 16 format = 'openapi' 17 18 def decode(self, bytes, **options): 19 """ 20 Takes a bytestring and returns a document. 21 """ 22 try: 23 data = json.loads(bytes.decode('utf-8')) 24 except ValueError as exc: 25 raise ParseError('Malformed JSON. %s' % exc) 26 27 base_url = options.get('base_url') 28 doc = _parse_document(data, base_url) 29 if not isinstance(doc, Document): 30 raise ParseError('Top level node must be a document.') 31 32 return doc 33 34 def encode(self, document, **options): 35 if not isinstance(document, Document): 36 raise TypeError('Expected a `coreapi.Document` instance') 37 data = generate_swagger_object(document) 38 return force_bytes(json.dumps(data)) 39