技术分享1 分钟阅读
Python 解析Swagger
Swagger是后端常用的接口管理工具,
import requests
url = 'https://cp-publgacademy.com/v2/api-docs'
def getswagger(url):
interfaces_list = []
rsp = requests.request("GET",url,verify=False).json()
paths = rsp['paths']
for path in paths:
interface = {
"description": "获取线索列表",
"url": "/track/list",
"method": "GET",
"header": "",
"par": "",
"data": "",
"module": ""
}
interface['url'] = path
for data in paths[path]:
interface['method'] = data.upper()
interface['description'] = paths[path][data]['summary']
interface['module'] = paths[path][data]['tags'][0]
if paths[path][data].get('parameters') != None:
parameters = paths[path][data].get('parameters')
for itme in parameters:
if itme.get('in') == 'body':
body_date = itme['schema']
interface['data'] = body_date
print(interface)
interfaces_list.append(interface)
print(interfaces_list)
return interfaces_list
getswagger(url)
swagger2解析为python请求
安装 pip install swagger2
import swagger2
url ='https://petstore.swagger.io/v2/swagger.json'
swagger = swagger2.parse(url,deep=5, verify=False)
print(json.dumps(swagger.apis))
源码中有些问题修改过后的文件 :
import re
import json
import urllib.parse
import uuid
import copy
class IncompleteModelError(BaseException):
pass
class Swagger:
def __init__(self,source,deep=5):
self.source = source
self.deep = deep
self.__scheme = self.schemes[0]
self.__host = self.source.get('host') or 'localhost'
def __property__(self,prop):
if isinstance(prop, dict):
_type = prop.get('type')
_format = prop.get('format')
_example = prop.get('example')
if _example:
prop = _example
elif _type == 'integer':
prop = 0
elif _type == 'number':
prop = 1.75
elif _type == 'boolean':
prop = True
elif _type == 'string':
if _format in [None,'byte', 'binary', 'password','int32','uuid']:
prop = 'string'
elif _format == 'date':
prop = '1970-01-01'
elif _format == 'date-time':
prop = '1970-01-01 00:00:00'
else:
prop = 'string'
elif _type == 'file':
prop = 'file.txt'
elif _type == 'array':
prop = []
elif _type == 'object':
prop = {}
else:
pass
return prop
@property
def schemes(self):
return self.source.get('schemes') or ['http']
@property
def scheme(self):
return self.__scheme
@scheme.setter
def scheme(self,value):
self.__scheme = value
@property
def host(self):
return self.__host
@host.setter
def host(self,value):
self.__host = value
@property
def basePath(self):
return self.source.get('basePath') or '/'
@property
def models(self):
ref_results = dict()
definitions = copy.deepcopy(self.source.get('definitions'))
for _ in range(self.deep):
for key in list(definitions.keys()):
properties = definitions[key].get('properties')
if properties:
ref_keys = re.findall(r'"#/definitions/(.+?)"',json.dumps(properties,ensure_ascii=False))
if ref_keys and len(ref_keys)>0:
# if set(ref_results.keys()).issuperset(set(ref_keys)):
for prop_name, prop in properties.items():
ref_key = prop.get('$ref')
if ref_key:
properties[prop_name] = ref_results.get(ref_key.split('/').pop())
else:
properties[prop_name] = self.__property__(prop)
ref_results.update({key: properties})
del definitions[key]
else:
for prop_name, prop in properties.items():
properties[prop_name] = self.__property__(prop)
ref_results.update({key: properties})
del definitions[key]
else:
ref_results.update({key: dict()})
del definitions[key]
# 无模型停止递归
if len(definitions) == 0:
break
if len(definitions) > 0:
raise IncompleteModelError('The model is incomplete. Please adjust the deep value of the swagger object.')
return ref_results
@property
def apis(self):
models = self.models
api_results = []
for path, forms in self.source.get('paths').items():
for method, form in forms.items():
name = form.get('summary')
contentType = form.get('consumes') or ['']
contentType = contentType[0]
_path = (self.basePath+path).replace('//','/')
url = urllib.parse.urlunparse((self.scheme, self.host, _path, None, None, None))
_api = {
'id': uuid.uuid4().hex,
'name': name,
'method': method,
'path': _path,
'url': url,
'headers':{'Content-Type': contentType},
'paths':dict(),
'query':dict(),
'json':dict(),
'form':dict(),
'formData':dict()
}
# 参数提取
if form.get('parameters') != None:
for param in form.get('parameters'):
_in = param.get('in')
paramName = param.get('name')
paramValue = self.__property__(param)
if isinstance(paramValue, dict):
schema = paramValue.get('schema')
if schema == None:
continue
elif schema.get('type') != None:
paramValue = self.__property__(schema)
else:
ref_key = re.search(r'"#/definitions/(.+?)"',json.dumps(schema,ensure_ascii=False))
if ref_key:
paramValue = models.get(ref_key.group(1))
if _in == 'header':
_api.get('headers').update({paramName: paramValue})
elif _in == 'path':
_api.get('paths').update({paramName: paramValue})
elif _in == 'query':
_api.get('query').update({paramName: paramValue})
elif _in == 'body':
_api.update({'json':paramValue})
elif _in == 'formData':
if 'multipart/form-data' in contentType:
_api.get('formData').update({paramName: paramValue})
else:
_api.get('form').update({paramName: paramValue})
api_results.append(_api)
return api_results
有关使用上的问题,欢迎您在底部评论区留言,一起交流~
读者评论
评论会同步写入该文在 Notion 中的页面底部(与正文同页,便于管理)。
暂无评论,欢迎抢沙发。