Notion Blog
技术分享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 中的页面底部(与正文同页,便于管理)。

0/1500

暂无评论,欢迎抢沙发。