requests接口POM模式封装设计

爱被打了一巴掌 2022-12-17 13:56 55阅读 0赞

以下没没有进行封装的代码：

import requests
    from base_api.base import Base
    
    
    class Tag(Base):
        _corpsecret = "ZuqJiGA6NjxfWJOizNlcoy0F0gEgIuGyAjHXbx3Cxyo"
    
        def __init__(self):
            self.get_token(self._corpsecret)
            self.data = self.api_load("../base_api/tag_api.yaml")
    
        def add(self, tagname):
            _url = 'https://qyapi.weixin.qq.com/cgi-bin/tag/create'
            r = requests.post(_url,
                              params={"access_token": self.get_token(self._corpsecret)},
                              json={"tagname": tagname})
            self.format(r)
            return r.json()
    
        def get(self, tagid):
            return self.api_send(self.data['get'])
            _url = 'https://qyapi.weixin.qq.com/cgi-bin/tag/get'
            r = requests.post(_url,
                              params={
                                  "access_token": self.get_token(self._corpsecret),
                                  "tagid": tagid,
                                      })
            self.format(r)
            return r.json()
    
        def delete(self, tagid):
            _url = 'https://qyapi.weixin.qq.com/cgi-bin/tag/delete'
            r = requests.get(_url,
                              params={
                                  "access_token": self.get_token(self._corpsecret),
                                  "tagid": tagid,
                                      })
            self.format(r)
            return r.json()

发现代码很多冗余，下面采取基于yaml数据驱动的形式来进行封装。

首先在base.py文件中：

import requests
    import json
    
    import yaml
    
    
    class Base:
        _token = []
        _token_url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
        _corpid = "wwcbefdd1de3927e42"
    
        @classmethod
        def yaml_load(cls, path):
            with open(path, encoding='utf-8') as f:
                return yaml.safe_load(f)
    
        @classmethod
        def format(cls, r):
            cls.r = r
            # print(json.dumps(r.json(), indent=2))
            print(json.dumps(json.loads(r.text), indent=2, ensure_ascii=False))
    
        def api_load(self, path):
            return self.yaml_load(path)
    
        def api_send(self, req: dict):
            req['params']['access_token'] = self._token[0]
            print(req)
            r = requests.request(req["method"],
                                 url=req["url"],
                                 params=req["params"],
                                 json=req["json"])
            self.format(r)
            return r.json()
    
        @classmethod
        def get_token(cls, corpsecret):
            # todo 这里还可以把时间记录下来，每次调用都判读一下时间
            if corpsecret not in cls._token:
                res_dict = cls.get_access_token(corpsecret)
                cls._token.append(res_dict["access_token"])
            return cls._token[-1]
    
        @classmethod
        def get_access_token(cls, corpsecret):
            _token_url_params = {
                "corpid": cls._corpid,
                "corpsecret": corpsecret,
            }
            res = requests.get(cls._token_url, params=_token_url_params)
            return res.json()

接下来的tag.py文件对get，post这些请求体的请求就变得简单了

from base_api.base import Base
    
    
    class Tag(Base):
        _corpsecret = "ZuqJiGA6NjxfWJOizNlcoy0F0gEgIuGyAjHXbx3Cxyo"
    
        def __init__(self):
            self.get_token(self._corpsecret)
            self.data = self.api_load("../base_api/tag_api.yaml")
    
        def get_api(self):
            pass
    
        def add(self):
            return self.api_send(self.data['add'])
    
        def get(self):
            return self.api_send(self.data['get'])
    
        def delete(self, tagid):
            return self.api_send(self.data['delete'])

接下来的用例层就更优雅了

from base_api.tag import Tag
    from base_api import base
    from base_api.group_chat import GroupChat
    import pytest
    
    
    class TestTag(object):
        # 测试数据的测试驱动
        data = base.Base.yaml_load("test_tag_data.yaml")
        @classmethod
        def setup_class(cls):
            cls.tag = Tag()
    
        def test_add(self, name):
            r = self.tag.add()
            assert r["errcode"] == 0
    
        def test_get(self):
            r = self.tag.get(3)
            assert r["errcode"] == 0
    
        def test_get_api(self):
            r = self.tag.get_api()
            assert r["errcode"] == 0
    
        def test_delete(self):
            r = self.tag.delete(3)
            assert r["errcode"] == 0

其中的数据被封装到test\_tag\_data.yaml文件中：

get:
      method: post
      url: https://qyapi.weixin.qq.com/cgi-bin/tag/get
      params:
        access_token: ${token}
        tagid: 3
      json:
    
    
    
    
    add:
      method: post
      url: https://qyapi.weixin.qq.com/cgi-bin/tag/create
      params:
        access_token: ${token}
        tagid: 3
      json:
        tagname: Tencent

以上就是基于步骤驱动的封装

代码：[https://github.com/chaunchengZeng/API-][https_github.com_chaunchengZeng_API-]

[https_github.com_chaunchengZeng_API-]: https://github.com/chaunchengZeng/API-