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

   1 #!/usr/bin/python
   2 # -*- coding: utf-8 -*-
   3 import urlparse
   4 import json
   5 import inspect
   6 import tornado.web
   7 import tornado.template
   8 from settings import SWAGGER_VERSION, URL_SWAGGER_API_LIST, URL_SWAGGER_API_SPEC, models
   9
  10 __author__ = 'serena'
  11
  12
  13 def json_dumps(obj, pretty=False):
  14     return json.dumps(obj, sort_keys=True, indent=4, separators=(',', ': ')) if pretty else json.dumps(obj)
  15
  16
  17 class SwaggerUIHandler(tornado.web.RequestHandler):
  18     def initialize(self, static_path, **kwds):
  19         self.static_path = static_path
  20
  21     def get_template_path(self):
  22         return self.static_path
  23
  24     def get(self):
  25         discovery_url = urlparse.urljoin(self.request.full_url(), self.reverse_url(URL_SWAGGER_API_LIST))
  26         self.render('index.html', discovery_url=discovery_url)
  27
  28
  29 class SwaggerResourcesHandler(tornado.web.RequestHandler):
  30     def initialize(self, api_version, exclude_namespaces, **kwds):
  31         self.api_version = api_version
  32         self.exclude_namespaces = exclude_namespaces
  33
  34     def get(self):
  35         self.set_header('content-type', 'application/json')
  36         u = urlparse.urlparse(self.request.full_url())
  37         resources = {
  38             'apiVersion': self.api_version,
  39             'swaggerVersion': SWAGGER_VERSION,
  40             'basePath': '%s://%s' % (u.scheme, u.netloc),
  41             'produces': ["application/json"],
  42             'description': 'Test Api Spec',
  43             'apis': [{
  44                 'path': self.reverse_url(URL_SWAGGER_API_SPEC),
  45                 'description': 'Test Api Spec'
  46             }]
  47         }
  48
  49         self.finish(json_dumps(resources, self.get_arguments('pretty')))
  50
  51
  52 class SwaggerApiHandler(tornado.web.RequestHandler):
  53     def initialize(self, api_version, base_url, **kwds):
  54         self.api_version = api_version
  55         self.base_url = base_url
  56
  57     def get(self):
  58         self.set_header('content-type', 'application/json')
  59         apis = self.find_api(self.application.handlers)
  60         if apis is None:
  61             raise tornado.web.HTTPError(404)
  62
  63         specs = {
  64             'apiVersion': self.api_version,
  65             'swaggerVersion': SWAGGER_VERSION,
  66             'basePath': urlparse.urljoin(self.request.full_url(), self.base_url)[:-1],
  67             'apis': [self.__get_api_spec__(path, spec, operations) for path, spec, operations in apis],
  68             'models': self.__get_models_spec(models)
  69         }
  70         self.finish(json_dumps(specs, self.get_arguments('pretty')))
  71
  72     def __get_models_spec(self, models):
  73         models_spec = {}
  74         for model in models:
  75             models_spec.setdefault(model.id, self.__get_model_spec(model))
  76         return models_spec
  77
  78     @staticmethod
  79     def __get_model_spec(model):
  80         return {
  81             'description': model.summary,
  82             'id': model.id,
  83             'notes': model.notes,
  84             'properties': model.properties,
  85             'required': model.required
  86         }
  87
  88     @staticmethod
  89     def __get_api_spec__(path, spec, operations):
  90         return {
  91             'path': path,
  92             'description': spec.handler_class.__doc__,
  93             'operations': [{
  94                 'httpMethod': api.func.__name__.upper(),
  95                 'nickname': api.nickname,
  96                 'parameters': api.params.values(),
  97                 'summary': api.summary,
  98                 'notes': api.notes,
  99                 'responseClass': api.responseClass,
 100                 'responseMessages': api.responseMessages,
 101             } for api in operations]
 102         }
 103
 104     @staticmethod
 105     def find_api(host_handlers):
 106         for host, handlers in host_handlers:
 107             for spec in handlers:
 108                 for (name, member) in inspect.getmembers(spec.handler_class):
 109                     if inspect.ismethod(member) and hasattr(member, 'rest_api'):
 110                         spec_path = spec._path % tuple(['{%s}' % arg for arg in member.rest_api.func_args])
 111                         operations = [member.rest_api for (name, member) in inspect.getmembers(spec.handler_class)
 112                                       if hasattr(member, 'rest_api')]
 113                         yield spec_path, spec, operations
 114                         break
 115
 116