utils/test/result_collection_api/tornado_swagger_ui/tornado_swagger/swagger.py

   1 #!/usr/bin/python
   2 # -*- coding: utf-8 -*-
   3
   4 import inspect
   5 from functools import wraps
   6 import epydoc.markup
   7 from HTMLParser import HTMLParser
   8 import tornado.web
   9 from settings import default_settings, models
  10 from handlers import swagger_handlers
  11
  12 __author__ = 'serena'
  13
  14
  15 class EpytextParser(HTMLParser):
  16     a_text = False
  17
  18     def __init__(self, tag):
  19         HTMLParser.__init__(self)
  20         self.tag = tag
  21         self.data = None
  22
  23     def handle_starttag(self, tag, attr):
  24         if tag == self.tag:
  25             self.a_text = True
  26
  27     def handle_endtag(self, tag):
  28         if tag == self.tag:
  29             self.a_text = False
  30
  31     def handle_data(self, data):
  32         if self.a_text:
  33             self.data = data
  34
  35     def get_data(self):
  36         return self.data
  37
  38
  39 class DocParser(object):
  40     def __init__(self):
  41         self.notes = None
  42         self.summary = None
  43         self.responseClass = None
  44         self.responseMessages = []
  45         self.params = {}
  46         self.properties = {}
  47
  48     def parse_docstring(self, text):
  49         if text is None:
  50             return
  51
  52         errors = []
  53         doc = epydoc.markup.parse(text, markup='epytext', errors=errors)
  54         _, fields = doc.split_fields(errors)
  55
  56         for field in fields:
  57             tag = field.tag()
  58             arg = field.arg()
  59             body = field.body()
  60             self._get_parser(tag)(arg=arg, body=body)
  61         return doc
  62
  63     def _get_parser(self, tag):
  64         parser = {
  65             'param': self._parse_param,
  66             'type': self._parse_type,
  67             'rtype': self._parse_rtype,
  68             'property': self._parse_property,
  69             'ptype': self._parse_ptype,
  70             'return': self._parse_return,
  71             'raise': self._parse_return,
  72             'notes': self._parse_notes,
  73             'description': self._parse_description,
  74         }
  75         return parser.get(tag, self._not_supported)
  76
  77     def _parse_param(self, **kwargs):
  78         arg = kwargs.get('arg', None)
  79         body = self._get_body(**kwargs)
  80         self.params.setdefault(arg, {}).update({
  81             'name': arg,
  82             'description': body,
  83             'paramType': arg,
  84             'required': True,
  85             'allowMultiple': False
  86         })
  87
  88         if 'paramType' not in self.params[arg]:
  89             self.params[arg]['paramType'] = 'query'
  90
  91     def _parse_type(self, **kwargs):
  92         arg = kwargs.get('arg', None)
  93         body = self._get_body(**kwargs)
  94         self.params.setdefault(arg, {}).update({
  95             'name': arg,
  96             'dataType': body
  97         })
  98
  99     def _parse_rtype(self, **kwargs):
 100         body = self._get_body(**kwargs)
 101         self.responseClass = body
 102
 103     def _parse_property(self, **kwargs):
 104         arg = kwargs.get('arg', None)
 105         self.properties.setdefault(arg, {}).update({
 106             'type': 'string'
 107         })
 108
 109     def _parse_ptype(self, **kwargs):
 110         arg = kwargs.get('arg', None)
 111         code = self._parse_epytext_para('code', **kwargs)
 112         link = self._parse_epytext_para('link', **kwargs)
 113         if code is None:
 114             self.properties.setdefault(arg, {}).update({
 115                 'type': link
 116            })
 117         elif code == 'list':
 118             self.properties.setdefault(arg, {}).update({
 119                 'type': 'array',
 120                 'items': {'type': link}
 121             })
 122
 123     def _parse_return(self, **kwargs):
 124         arg = kwargs.get('arg', None)
 125         body = self._get_body(**kwargs)
 126         self.responseMessages.append({
 127             'code': arg,
 128             'message': body
 129         })
 130
 131     def _parse_notes(self, **kwargs):
 132         body = self._get_body(**kwargs)
 133         self.notes = self._sanitize_doc(body)
 134
 135     def _parse_description(self, **kwargs):
 136         body = self._get_body(**kwargs)
 137         self.summary = self._sanitize_doc(body)
 138
 139     def _not_supported(self, **kwargs):
 140         pass
 141
 142     @staticmethod
 143     def _sanitize_doc(comment):
 144         return comment.replace('\n', '<br/>') if comment else comment
 145
 146     @staticmethod
 147     def _get_body(**kwargs):
 148         body = kwargs.get('body', None)
 149         return body.to_plaintext(None).strip() if body else body
 150
 151     @staticmethod
 152     def _parse_epytext_para(tag, **kwargs):
 153         def _parse_epytext(tag, body):
 154             epytextParser = EpytextParser(tag)
 155             epytextParser.feed(str(body))
 156             data = epytextParser.get_data()
 157             epytextParser.close()
 158             return data
 159
 160         body = kwargs.get('body', None)
 161         return _parse_epytext(tag, body) if body else body
 162
 163
 164 class model(DocParser):
 165     def __init__(self, cls=None, *args, **kwargs):
 166         super(model, self).__init__()
 167         self.id = cls.__name__
 168         self.args = args
 169         self.kwargs = kwargs
 170         self.required = []
 171
 172         if '__init__' in dir(cls):
 173             self._parse_args(cls.__init__)
 174         self.parse_docstring(inspect.getdoc(cls))
 175         models.append(self)
 176
 177     def _parse_args(self, func):
 178         argspec = inspect.getargspec(func)
 179         argspec.args.remove("self")
 180         defaults = {}
 181         if argspec.defaults:
 182             defaults = list(zip(argspec.args[-len(argspec.defaults):], argspec.defaults))
 183         required_args_count = len(argspec.args) - len(defaults)
 184         for arg in argspec.args[:required_args_count]:
 185             self.required.append(arg)
 186             self.properties.setdefault(arg, {'type': 'string'})
 187         for arg, default in defaults:
 188             self.properties.setdefault(arg, {'type': 'string', "default": default})
 189
 190
 191
 192 class operation(DocParser):
 193     def __init__(self, nickname=None, **kwds):
 194         super(operation, self).__init__()
 195         self.nickname = nickname
 196         self.func = None
 197         self.func_args = []
 198         self.kwds = kwds
 199
 200     def __call__(self, *args, **kwds):
 201         if self.func:
 202             return self.func(*args, **kwds)
 203
 204         func = args[0]
 205         self._parse_operation(func)
 206
 207         @wraps(func)
 208         def __wrapper__(*in_args, **in_kwds):
 209             return self.func(*in_args, **in_kwds)
 210
 211         __wrapper__.rest_api = self
 212         return __wrapper__
 213
 214     def _parse_operation(self, func):
 215         self.func = func
 216
 217         self.__name__ = func.__name__
 218         self._parse_args(func)
 219         self.parse_docstring(inspect.getdoc(self.func))
 220
 221     def _parse_args(self, func):
 222         argspec = inspect.getargspec(func)
 223         argspec.args.remove("self")
 224
 225         defaults = []
 226         if argspec.defaults:
 227             defaults = argspec.args[-len(argspec.defaults):]
 228
 229         for arg in argspec.args:
 230             if arg in defaults:
 231                 required = False
 232             else:
 233                 required = True
 234             self.params.setdefault(arg, {
 235                 'name': arg,
 236                 'required': required,
 237                 'paramType': 'path',
 238                 'dataType': 'string'
 239             })
 240         self.func_args = argspec.args
 241
 242
 243 def docs(**opts):
 244     default_settings.update(opts)
 245
 246
 247 class Application(tornado.web.Application):
 248     def __init__(self, handlers=None, default_host="", transforms=None, **settings):
 249         super(Application, self).__init__(swagger_handlers() + handlers, default_host, transforms, **settings)