解析和增量导出 Swagger 的 JSON 文件

凌云 关注

收藏于 : 2018-11-13 09:44   被转藏 : 1   

用了Swagger的程序在运行时会扫描类和方法里的注解,生成JSON文档,默认地址为接口访问地址后面加 /v2/api-docs

本地启动的服务默认为 http://localhost:8080/v2/api-docs

Swagger UI生成的网页的数据源就是这JSON文件

缺点:

  • 只有服务启动了才能查看,没有离线版
  • 只支持全量导出,有些由于历史遗留问题没细分的服务可能有几千个接口,不方便查看
  • 靠Swagger配置类里设的分组难以满足“只看这版本新增接口”这样的需求

思路

写脚本解析Swagger生成的json文件,可以干很多事:

  • 对比新旧2份文件,分别导出新增、删除的接口
  • 按关键字导出特定接口
  • 修改页面发请求时访问的服务器地址
  • ……

json文件格式如下,只列出一定会出现的属性:

                  {   "swagger"   :   "2.0"   ,   "host"   :   "xxx"   ,   "basePath"   :   "/"   ,   "tags"   :[   {   "name"   :   "xxx-controller"   ,   "description"   :   "xxx"   },   ...   ],   "paths"   :   {   "<接口地址1>"   :   {   ...   },   "<接口地址2>"   :   {   ...   },   ...   },   "definitions"   :   {   "<实体类1>"   :   {   ...   },   "<实体类2>"   :   {   ...   },   ...   }   }   
  • host :服务器域名或IP
  • basePath :必须以 / 开头,访问某服务的接口时网关里设的那个前缀。如果host指定端口直连,这里为 /
  • tags :默认为Controller类名( @Api 注解了的类)。建议保持默认,不需要设 @Api(tags={"xx", "yy"}) @Api(value="xxx")
  • paths :接口( @ApiOperation 注解了的方法)
  • definitions :实体类( @ApiModel 注解了的类)

其中host和basePath最重要,这里设置得不对,页面上就没法访问到服务


以下脚本原本打算用于Confluence的Open API插件,把json导出来之后粘贴到插件里就有跟swagger ui一样的效果


但那插件非常不方便,后来有了 这帖 里的集中展示之后就不用那插件了,export_apis 和 export_models 这2个方法也显得多余了(当时想着万一还想加1个就粘贴文件里的内容插进去)


实现

新建Python文件,比如叫做 swagger_json_dump.py

再另外用Java搭个Spring Boot + Swagger的demo,看看不同设置下的输出,边调边写,都是体力活

                  #   coding   =   utf   -   8   import   argparse   import   errno   import   json   import   os   import   re   import   sys   from   datetime   import   datetime   from   os   import   path   import   requests   HOST   =   '   localhost:   8080   '   BASE_PATH   =   '/'   def   check_python_version   ():   if   sys   .   version_info   [   0   ]   <   3   or   (   sys   .   version_info   [   0   ]   ==   3   and   sys   .   version_info   [   1   ]   <   2   ):   print   (   '   Must   use   Python   3.2   +.   Current   version:   '   ,   sys   .   version   )   sys   .   exit   (   1   )   def   parse_cmdline_args   ():   parser   =   argparse   .   ArgumentParser   ()   parser   .   add_argument   (   '   --   url   '   ,   '   -   u   '   ,   help   =   '   Swagger   API   docs   URL   .   Format:   "http://<host>[:<port>][/<basePath>]/v2/api-docs"   (   default   :   http:   //localhost:8080/v2/api-docs)')   parser   .   add_argument   (   '   --   host   '   ,   '   -   H   '   ,   help   =   '   Domain   name   or   IP   .   (   default   :   "localhost:8080"   )   '   )   parser   .   add_argument   (   '   --   basePath   '   ,   '   -   b   '   ,   help   =   '   URL   prefix   for   all   API   paths   ,   relative   to   the   host   root   .   It   must   start   with   a   leading   slash   "/"   .   (   default   :   "/"   )   '   )   parser   .   add_argument   (   '   --   outputDir   '   ,   '   -   o   '   ,   help   =   '   Output   directory   path   .   (   default   :   current   working   directory   )   '   )   parser   .   add_argument   (   '   --   compare   '   ,   '   -   c   '   ,   help   =   '   Compare   to   another   (   old   )   Swagger   JSON   file   and   export   newly   added   APIs   and   models   .   '   )   parser   .   add_argument   (   '   --   api   '   ,   '   -   a   '   ,   help   =   '   Export   specified   API   (   s   ).   Comma   -   separated   (   partial   )   API   path   (   s   ),   case   sensitive   .   eg   .   "register,login"   '   )   parser   .   add_argument   (   '   --   model   '   ,   '   -   m   '   ,   help   =   '   Export   specified   model   (   s   ).   Comma   -   separated   (   partial   )   model   name   (   s   ),   case   sensitive   .   eg   .   "User,Files"   '   )   return   parser   .   parse_args   ()   def   get_swagger_json   (   url   ):   try   :   r   =   requests   .   get   (   url   )   except:   raise   Exception   (   '   [   ERROR   ]   Invalid   URL   or   service   is   not   running   .   URL:   {}   '   .   format   (   url   ))   if   r   .   status_code   ==   200   :   try   :   response   =   r   .   json   ()   if   response   .   get   (   '   swagger   '   ):   return   response   else   :   raise   Exception   (   '   [   ERROR   ]   Not   a   Swagger   JSON   file:   \   n   '   +   str   (   response   ))   except:   raise   else   :   raise   Exception   (   '   [   ERROR   ]   Cannot   get   JSON   .   Status   code:   {},   content:   {}   '   .   format   (   r   .   status_code   ,   r   .   text   ))   def   make_dir   (   dir_path   ):   if   not   path   .   exists   (   dir_path   ):   try   :   os   .   mkdir   (   dir_path   )   except   OSError   as   e:   if   e   .   errno   !=   errno   .   EEXIST   :   raise   def   export_whole_json   (   swagger_json   ,   output_path   ):   make_dir   (   output_path   )   with   open   (   path   .   join   (   output_path   ,   '   all   .   json   '   ),   'w'   ,   encoding   =   '   utf   -   8   '   )   as   f:   json   .   dump   (   swagger_json   ,   f   ,   ensure_ascii   =   False   )   def   export_apis   (   swagger_json   ,   output_path   ):   make_dir   (   output_path   )   api_dir   =   path   .   join   (   output_path   ,   '   api   '   )   make_dir   (   api_dir   )   tags   =   swagger_json   [   '   tags   '   ]   for   tag   in   tags:   tag_name   =   tag   [   '   name   '   ]   tag_dir   =   path   .   join   (   api_dir   ,   tag_name   )   make_dir   (   tag_dir   )   apis   =   swagger_json   [   '   paths   '   ]   for   api   in   apis:   if   tag_name   in   json   .   dumps   (   apis   [   api   ],   ensure_ascii   =   False   ):   api_file   =   path   .   join   (   tag_dir   ,   api   .   replace   (   '/'   ,   '   --   '   )   +   '   .   json   '   )   with   open   (   path   .   join   (   tag_dir   ,   api_file   ),   'w'   ,   encoding   =   '   utf   -   8   '   )   as   f:   json   .   dump   (   apis   [   api   ],   f   ,   ensure_ascii   =   False   )   def   export_models   (   swagger_json   ,   output_path   ):   make_dir   (   output_path   )   model_dir   =   path   .   join   (   output_path   ,   '   model   '   )   make_dir   (   model_dir   )   models   =   swagger_json   [   '   definitions   '   ]   for   model   in   models:   with   open   (   path   .   join   (   model_dir   ,   model   +   '   .   json   '   ),   'w'   ,   encoding   =   '   utf   -   8   '   )   as   f:   json   .   dump   (   models   [   model   ],   f   ,   ensure_ascii   =   False   )   def   search_refs   (   string   ):   return   re   .   findall   (   '   "\$ref"   :   "#/definitions/(.+?)"   '   ,   string   )   def   write_selected_elements_to_file   (   selections   ,   elements   ,   file_obj   ):   refs   =   []   elements_to_write   =   []   for   selection   in   set   (   selections   ):   for   element   in   elements:   if   selection   in   element:   content   =   json   .   dumps   (   elements   [   element   ],   ensure_ascii   =   False   )   elements_to_write   .   append   (   '\   t   \   t   "{}"   :   {},   \   n   '   .   format   (   element   ,   content   ))   refs   .   extend   (   search_refs   (   content   ))   for   element   in   set   (   elements_to_write   ):   file_obj   .   write   (   element   )   return   list   (   set   (   refs   ))   def   write_all_ref_models_to_file   (   ref_list   ,   model_list   ,   file_obj   ):   ref_list   =   list   (   set   (   ref_list   ))   models_to_write   =   []   while   len   (   ref_list   ):   for   ref   in   set   (   ref_list   ):   if   ref   in   model_list:   content   =   json   .   dumps   (   model_list   [   ref   ],   ensure_ascii   =   False   )   models_to_write   .   append   (   '\   t   \   t   "{}"   :   {},   \   n   '   .   format   (   ref   ,   content   ))   ref_list   .   extend   (   search_refs   (   content   ))   if   ref   in   ref_list:   ref_list   .   remove   (   ref   )   for   model   in   set   (   models_to_write   ):   file_obj   .   write   (   model   )   def   export_selected_docs   (   swagger_json   ,   output_file   ,   api_list   =   None   ,   model_list   =   None   ):   if   not   api_list:   api_list   =   []   if   not   model_list:   model_list   =   []   apis   =   swagger_json   [   '   paths   '   ]   models   =   swagger_json   [   '   definitions   '   ]   refs   =   []   with   open   (   output_file   ,   'w'   ,   encoding   =   '   utf   -   8   '   )   as   f:   f   .   write   (   '   {   \   n   \   t   "swagger"   :   "2.0"   ,   \   n   '   )   f   .   write   (   '\   t   "info"   :   {},   \   n   '   .   format   (   json   .   dumps   (   swagger_json   [   '   info   '   ],   ensure_ascii   =   False   )))   f   .   write   (   '\   t   "host"   :   "{}"   ,   \   n   '   .   format   (   HOST   ))   f   .   write   (   '\   t   "basePath"   :   "{}"   ,   \   n   '   .   format   (   BASE_PATH   ))   f   .   write   (   '\   t   "tags"   :   {},   \   n   '   .   format   (   json   .   dumps   (   swagger_json   [   '   tags   '   ],   ensure_ascii   =   False   )))   f   .   write   (   '\   t   "paths"   :   {   \   n   '   )   refs   .   extend   (   write_selected_elements_to_file   (   api_list   ,   apis   ,   f   ))   f   .   write   (   '\   t   \   t   ""   :{}   \   n   '   )   #   Use   empty   element   to   avoid   trailing   comma   issue   f   .   write   (   '\   t   },   \   n   '   )   f   .   write   (   '\   t   "definitions"   :   {   \   n   '   )   refs   .   extend   (   write_selected_elements_to_file   (   model_list   ,   models   ,   f   ))   write_all_ref_models_to_file   (   refs   ,   models   ,   f   )   f   .   write   (   '\   t   \   t   ""   :{}   \   n   '   )   #   Use   empty   element   to   avoid   trailing   comma   issue   f   .   write   (   '\   t   }   \   n   '   )   f   .   write   (   '}'   )   def   export_diff_docs   (   old_file_path   ,   output_path   ):   if   path   .   exists   (   old_file_path   )   and   path   .   isfile   (   old_file_path   ):   with   open   (   old_file_path   ,   encoding   =   '   utf   -   8   '   )   as   f:   old_swagger_json   =   json   .   load   (   f   )   new_apis   =   set   (   j   [   '   paths   '   ].   keys   ())   old_apis   =   set   (   old_swagger_json   [   '   paths   '   ].   keys   ())   added   =   new_apis   -   old_apis   missing   =   old_apis   -   new_apis   print   (   '\   nAdded   APIs:   \   n   {}   '   .   format   (   added   ))   print   (   '   Missing   APIs:   \   n   {}   '   .   format   (   missing   ))   if   len   (   added   ):   output_added   =   path   .   join   (   output_path   ,   '   diff   -   added   .   json   '   )   export_selected_docs   (   j   ,   output_added   ,   added   )   print   (   '\   nAdded   APIs   are   saved   to:   \   n   {}   '   .   format   (   output_added   ))   else   :   print   (   '   [   INFO   ]   No   new   APIs   found   .   '   )   if   len   (   missing   ):   output_missing   =   path   .   join   (   output_path   ,   '   diff   -   missing   .   json   '   )   export_selected_docs   (   old_swagger_json   ,   output_missing   ,   missing   )   print   (   '\   nMissing   APIs   are   saved   to:   \   n   {}   '   .   format   (   output_missing   ))   else   :   print   (   '   [   ERROR   ]   No   such   file:   {}   '   .   format   (   old_file_path   ))   if   __name__   ==   '   __main__   '   :   check_python_version   ()   args   =   parse_cmdline_args   ()   if   args   .   url   :   match   =   re   .   search   (   '   https   ?:   //(.+?)(?:/v2/api-docs.*?)', args.url)   if   match:   HOST   =   match   .   groups   ()[   0   ]   if   '/'   in   HOST:   match2   =   re   .   search   (   '   (.+?)(/.*)   '   ,   HOST   )   if   match2:   HOST   =   match2   .   groups   ()[   0   ]   BASE_PATH   =   match2   .   groups   ()[   1   ]   #   overwrite   "host"   and   "basePath"   in   Swagger   JSON   if   args   .   host   :   HOST   =   args   .   host   if   args   .   basePath   :   BASE_PATH   =   args   .   basePath   swagger_url   =   args   .   url   or   '   http:   //{}{}/v2/api-docs'.format(HOST, BASE_PATH)   output_dir   =   path   .   abspath   (   path   .   expanduser   (   args   .   outputDir   ))   if   args   .   outputDir   else   path   .   join   (   os   .   getcwd   (),   '   swagger_json_docs   -{}   '   .   format   (   datetime   .   now   ().   strftime   (   '   %   Y   %   m   %   d   -%   H   %   M   %   S   '   )))   j   =   get_swagger_json   (   swagger_url   )   export_whole_json   (   j   ,   output_dir   )   export_apis   (   j   ,   output_dir   )   export_models   (   j   ,   output_dir   )   print   (   '   Saved   to:   \   n   {}   '   .   format   (   output_dir   ))   if   args   .   compare   :   old_file   =   args   .   compare   export_diff_docs   (   old_file   ,   output_dir   )   if   args   .   api   or   args   .   model   :   selected_apis   =   args   .   api   .   strip   ().   split   (   ','   )   if   args   .   api   else   []   selected_models   =   args   .   model   .   strip   ().   split   (   ','   )   if   args   .   model   else   []   output_file   =   path   .   join   (   output_dir   ,   '   selected   .   json   '   )   export_selected_docs   (   j   ,   output_file   ,   selected_apis   ,   selected_models   )   print   (   '\   nSelected   APIs   and   models   are   saved   to:   \   n   {}   '   .   format   (   output_file   ))   

注意插了 "":{} 作为 paths definitions 数组的最后1个元素

因为JSON不允许数组和对象最后1项后面带逗号,又不知有多少项,插个空元素避免问题,基本不影响页面展示


用法

需要Python 3.2+

查看帮助

                  python3   swagger_json_dump   .   py   --   help   

连本地服务

                  python3   swagger_json_dump   .   py   #   默认从   http:   //localhost:8080/v2/api-docs 获取json文件   

默认在当前工作目录生成名为 swagger_json_docs-<年月日>-<时分秒> 的文件夹

  • 整个json的内容原样保存到 all.json
  • 实体类相关的内容($.definitions.<类名> 的值)拆分保存在 model/<类名>.json 下
  • 接口相关的内容($.paths.<接口地址> 的值)拆分保存在 api/<--path--to–api>.json 下(文件名里不能有/,用 -- 代替)

指定swagger url

                  python3   swagger_json_dump   .   py   --   url   "http://localhost:8080/v2/api-docs?group=user"   python3   swagger_json_dump   .   py   --   url   "http://192.168.3.231/faq/v2/api-docs"   

如果swagger返回json的地址不是默认的 /v2/api-docs (例如在配置类里设了分组,要在后面加 ?group=XXX

又或者服务不在本地等等,都要指定url

覆盖输出文档里的 host basePath 属性

                  python3   swagger_json_dump   .   py   --   url   "http://localhost:8080/v2/api-docs"   --   host   "192.168.3.231"   --   basePath   "/faq"   

这2个属性决定swagger生成的网页做接口测试时连什么地址

脚本默认从url里提取,如果url和目标服务器不一样,需要手动指定

(例如上面url指向本地服务,但想接口写好之后让人在开发环境调试)

                  python3   swagger_json_dump   .   py   --   host   "192.168.3.231"   --   basePath   "/faq"   #    http:   //192.168.3.231/faq/v2/api-docs 获取json文件   

如果使用默认地址、无分组,只是服务器不一样,可以只指定这2个来代替输完整url

指定输出目录

                  python3   swagger_json_dump   .   py   --   output   "~/apidoc"   

找出改动过的接口

                  python3   swagger_json_dump   .   py   --   compare   "<之前的json文件路径>"   

从线上拿最新的json跟保存的文件比较,如果有改动,新增的接口保存到 diff-added.json,删除的接口保存到 diff-missing.json

接口用到的数据模型都会加进文件里

导出部分接口

                  python3   swagger_json_dump   .   py   --   api   "/collect"   python3   swagger_json_dump   .   py   --   api   "findByLabelName,findByUserId"   

指定完整或部分地址,区分大小写,多个关键字用逗号分隔

保存到 selected.json,引用到的模型也包含在里面

导出部分模型

                  python3   swagger_json_dump   .   py   --   model   "User,Files"   

同样保存到 selected.json,不指定导出的接口时,文件里只有模型没有接口


我不懂 Python,以上全是我编的,我实在编不下去了……

 阅读文章全部内容  
点击查看
文章点评
相关文章
凌云 关注

文章收藏:9037

TA的最新收藏