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 import tornado.web
   8 from settings import default_settings, models
   9 from handlers import swagger_handlers
  10
  11 __author__ = 'serena'
  12
  13
  14 class DocParser(object):
  15     def __init__(self):
  16         self.notes = None
  17         self.summary = None
  18         self.responseClass = None
  19         self.responseMessages = []
  20         self.params = {}
  21         self.properties = {}
  22
  23     def parse_docstring(self, text):
  24         if text is None:
  25             return
  26
  27         errors = []
  28         doc = epydoc.markup.parse(text, markup='epytext', errors=errors)
  29         _, fields = doc.split_fields(errors)
  30
  31         for field in fields:
  32             tag = field.tag()
  33             arg = field.arg()
  34             body = field.body().to_plaintext(None).strip()
  35             self._get_parser(tag)(arg=arg, body=body)
  36         return doc
  37
  38     def _get_parser(self, tag):
  39         parser = {
  40             'param': self._parse_param,
  41             'type': self._parse_type,
  42             'rtype': self._parse_rtype,
  43             'property': self._parse_property,
  44             'ptype': self._parse_ptype,
  45             'return': self._parse_return,
  46             'raise': self._parse_return,
  47             'notes': self._parse_notes,
  48             'description': self._parse_description,
  49         }
  50         return parser.get(tag, self._not_supported)
  51
  52     def _parse_param(self, **kwargs):
  53         arg = kwargs.get('arg', None)
  54         body = kwargs.get('body', None)
  55         self.params.setdefault(arg, {}).update({
  56             'name': arg,
  57             'description': body,
  58             'paramType': arg,
  59             'required': True,
  60             'allowMultiple': False
  61         })
  62
  63         if 'paramType' not in self.params[arg]:
  64             self.params[arg]['paramType'] = 'query'
  65
  66     def _parse_type(self, **kwargs):
  67         arg = kwargs.get('arg', None)
  68         body = kwargs.get('body', None)
  69         self.params.setdefault(arg, {}).update({
  70             'name': arg,
  71             'dataType': body
  72         })
  73
  74     def _parse_rtype(self, **kwargs):
  75         body = kwargs.get('body', None)
  76         self.responseClass = body
  77
  78     def _parse_property(self, **kwargs):
  79         arg = kwargs.get('arg', None)
  80         self.properties.setdefault(arg, {}).update({
  81             'type': 'string'
  82         })
  83
  84     def _parse_ptype(self, **kwargs):
  85         arg = kwargs.get('arg', None)
  86         body = kwargs.get('body', None)
  87         self.properties.setdefault(arg, {}).update({
  88             'type': body
  89         })
  90
  91     def _parse_return(self, **kwargs):
  92         arg = kwargs.get('arg', None)
  93         body = kwargs.get('body', None)
  94         self.responseMessages.append({
  95             'code': arg,
  96             'message': body
  97         })
  98
  99     def _parse_notes(self, **kwargs):
 100         body = kwargs.get('body', '')
 101         self.notes = self._sanitize_doc(body)
 102
 103     def _parse_description(self, **kwargs):
 104         body = kwargs.get('body', '')
 105         self.summary = self._sanitize_doc(body)
 106
 107     def _not_supported(self, **kwargs):
 108         pass
 109
 110     @staticmethod
 111     def _sanitize_doc(comment):
 112         return comment.replace('\n', '<br/>') if comment else comment
 113
 114
 115 class model(DocParser):
 116     def __init__(self, cls=None, *args, **kwargs):
 117         super(model, self).__init__()
 118         self.id = cls.__name__
 119         self.args = args
 120         self.kwargs = kwargs
 121         self.required = []
 122
 123         if '__init__' in dir(cls):
 124             self._parse_args(cls.__init__)
 125         self.parse_docstring(inspect.getdoc(cls))
 126         models.append(self)
 127
 128     def _parse_args(self, func):
 129         argspec = inspect.getargspec(func)
 130         argspec.args.remove("self")
 131         defaults = {}
 132         if argspec.defaults:
 133             defaults = list(zip(argspec.args[-len(argspec.defaults):], argspec.defaults))
 134         required_args_count = len(argspec.args) - len(defaults)
 135         for arg in argspec.args[:required_args_count]:
 136             self.required.append(arg)
 137             self.properties.setdefault(arg, {'type': 'string'})
 138         for arg, default in defaults:
 139             self.properties.setdefault(arg, {'type': 'string', "default": default})
 140
 141
 142
 143 class operation(DocParser):
 144     def __init__(self, nickname=None, **kwds):
 145         super(operation, self).__init__()
 146         self.nickname = nickname
 147         self.func = None
 148         self.func_args = []
 149         self.kwds = kwds
 150
 151     def __call__(self, *args, **kwds):
 152         if self.func:
 153             return self.func(*args, **kwds)
 154
 155         func = args[0]
 156         self._parse_operation(func)
 157
 158         @wraps(func)
 159         def __wrapper__(*in_args, **in_kwds):
 160             return self.func(*in_args, **in_kwds)
 161
 162         __wrapper__.rest_api = self
 163         return __wrapper__
 164
 165     def _parse_operation(self, func):
 166         self.func = func
 167
 168         self.__name__ = func.__name__
 169         self._parse_args(func)
 170         self.parse_docstring(inspect.getdoc(self.func))
 171
 172     def _parse_args(self, func):
 173         argspec = inspect.getargspec(func)
 174         argspec.args.remove("self")
 175
 176         defaults = []
 177         if argspec.defaults:
 178             defaults = argspec.args[-len(argspec.defaults):]
 179
 180         for arg in argspec.args:
 181             if arg in defaults:
 182                 required = False
 183             else:
 184                 required = True
 185             self.params.setdefault(arg, {
 186                 'name': arg,
 187                 'required': required,
 188                 'paramType': 'path',
 189                 'dataType': 'string'
 190             })
 191         self.func_args = argspec.args
 192
 193
 194 def docs(**opts):
 195     default_settings.update(opts)
 196
 197
 198 class Application(tornado.web.Application):
 199     def __init__(self, handlers=None, default_host="", transforms=None, **settings):
 200         super(Application, self).__init__(swagger_handlers() + handlers, default_host, transforms, **settings)