使用wxpython开发跨平台桌面应用，对WebAPI调用接口的封装

我在前面介绍的系统界面功能，包括菜单工具栏、业务表的数据，开始的时候，都是基于模拟的数据进行测试，数据采用JSON格式处理，通过辅助类的方式模拟实现数据的加载及处理，这在开发初期是一个比较好的测试方式，不过实际业务的数据肯定是来自后端，包括本地数据库，SqlServer、Mysql、Oracle、Sqlite、PostgreSQL等，或者后端的WebAPI接口获取，本篇随笔逐步介绍如何对后端的数据接口进行建模以及提供本地WebAPI代理接口类的处理过程。

1、定义Web API接口类并测试API调用基类

我在随笔《
使用wxpython开发跨平台桌面应用，动态工具的创建处理
》中介绍了关于工具栏和菜单栏的数据类，以及模拟方式获得数据进行展示，如下界面所示。

如菜单数据的类信息，如下所示。

classMenuInfo:

    id: str#菜单ID
    pid: str  #父菜单ID
    label: str  #菜单名称
    icon: str = None  #菜单图标
    path: str = None  #菜单路径，用来定位视图
    tips: str = None  #菜单提示
    children: list["MenuInfo"] = None

这些数据和后端数据接口的定义一致，那么就很容易切换到动态的接口上。

在系统开发的初期，我们可以先试用模拟方式获得数据集合，如通过一个工具来来获得数据，如下所示。

为了更好的符合实际的业务需求，我们往往需要根据服务端的接口定义来定义调用Web API接口的信息。

我们为了全部采用Python语言进行开发，包括后端的内容，采用
基于SqlAlchemy+Pydantic+FastApi
的后端框架

该后端接口采用统一的接口协议，标准协议如下所示。

{"success": false,"result":  T ,"targetUrl": "string","UnAuthorizedRequest": false,"errorInfo": {"code": 0,"message": "string","details": "string"}

}

其中的result是我们的数据返回，有可能是基本类型（如字符串、数值、布尔型等），也有可能是类集合，对象信息，字典信息等等。

如果是分页查询返回结果集合，其结果如下所示。

展开单条记录明细如下所示。

如果我们基于Pydantic模型定义，我们的Python对象类定义代码如下所示

from pydantic importBaseModelfrom typing importGeneric, Type, TypeVar, Optional

T= TypeVar("T")#自定义返回模型-统一返回结果
classAjaxResponse(BaseModel, Generic[T]):

    success: bool=False

    result: Optional[T]=None

    targetUrl: Optional[str]=None

    UnAuthorizedRequest: Optional[bool]=False

    errorInfo: Optional[ErrorInfo]= None

也就是结合泛型的方式，这样定义可以很好的抽象不同的业务类接口到基类BaseApi中，这样增删改查等处理的接口都可以抽象到BaseApi里面了。

权限模块我们涉及到的用户管理、机构管理、角色管理、菜单管理、功能管理、操作日志、登录日志等业务类，那么这些类继承BaseApi，就会具有相关的接口了，如下所示继承关系。

2、对异步调用进行测试和接口封装

为了理解客户端Api类的处理，我们先来介绍一些简单的pydantic 入门处理，如下我们先定义一些实体类用来承载数据信息，如下所示。

from typing importList, TypeVar, Optional, Generic, Dict, Anyfrom datetime importdatetimefrom pydantic importBaseModel, Field

T= TypeVar("T")classAjaxResult(BaseModel, Generic[T]):"""测试统一接口返回格式"""success: bool=True

    message: Optional[str]=None

    result: Optional[T]=NoneclassPagedResult(BaseModel, Generic[T]):"""分页查询结果"""total: int

    items: List[T]classCustomer(BaseModel):"""客户信息类"""name: str

    age: int

一般业务的结果是对应的记录列表，或者实体类对象格式，我们先来测试解析下它们的JSON数据，有助于我们理解。

#对返回结果数据格式的处理
json_data = """{

    "total": 100,

    "items": [

        {"name": "Alice", "age": 25},

        {"name": "Bob", "age": 30},

        {"name": "Charlie", "age": 35}

    ]

}"""paged_result=PagedResult.model_validate_json(json_data)print(paged_result.total)print(paged_result.items)

以上正常解析到数据，输出结果如下所示。

100[{'name': 'Alice', 'age': 25}, {'name': 'Bob', 'age': 30}, {'name': 'Charlie', 'age': 35}]

True

如果我们换为统一返回的结果进行测试，如下所示。

json_data = """{

    "success": true,

    "message": "success",

    "result": {

        "total": 100,

        "items": [

            {"name": "Alice", "age": 25},

            {"name": "Bob", "age": 30},

            {"name": "Charlie", "age": 35}

        ]

    }

}"""ajax_result=AjaxResult[PagedResult].model_validate_json(json_data)print(ajax_result.success)print(ajax_result.message)print(ajax_result.result.total)print(ajax_result.result.items)

同样的可以获得正常的输出。

True

success100[{'name': 'Alice', 'age': 25}, {'name': 'Bob', 'age': 30}, {'name': 'Charlie', 'age': 35}]

我们通过 model_validate_json 接口可以转换字符串内容为对应的业务类对象，而通过 model_validate 函数可以转换JSON格式为业务类对象。

而对于接口的继承处理，我们采用了泛型的处理，可以极大的减少基类代码的编写，如下基类定义和子类定义，就可以简单很多，所有逻辑放在基类处理即可。

classBaseApi(Generic[T]):def test(self) ->AjaxResult[Dict[str, Any]]:

        json_data= """{

            "success": true,

            "message": "success",

            "result": {"name": "Alice", "age": 25}

        }"""result=AjaxResult[Dict[str, Any]].model_validate_json(json_data)returnresultdef get(self, id: int) ->AjaxResult[T]:

        json_data= """{

            "success": true,

            "message": "success",

            "result": {"name": "Alice", "age": 25}

        }"""result=AjaxResult[T].model_validate_json(json_data)returnresultdef getlist(self) ->AjaxResult[List[T]]:

        json_data= """{

            "success": true,

            "message": "success",

            "result": [

                {"name": "Alice", "age": 25},

                {"name": "Bob", "age": 30},

                {"name": "Charlie", "age": 35}

            ]

        }"""result=AjaxResult[List[T]].model_validate_json(json_data)returnresultclass UserApi(BaseApi[Customer]):passuser_api=UserApi()

result=user_api.getlist()print(result.success)print(result.message)print(result.result)



result= user_api.get(1)print(result.success)print(result.message)print(result.result)



result=user_api.test()print(result.success)print(result.message)print(result.result)

可以看到，子类只需要明确好继承关系即可，不需要编写任何多余的代码，但是又有了具体的接口处理。

3、实际HTTTP请求的封装处理

一般对于服务端接口的处理，我们可能需要引入 aiohttp 来处理请求，并结合Pydantic的模型处理，是的数据能够正常的转换，和上面的处理方式一样。

首先我们需要定义一个通用HTTP请求的类来处理常规的HTTP接口数据的返回，如下所示。

classApiClient:

    _access_token= None  #类变量，用于全局共享 access_token
@classmethoddefset_access_token(cls, token):"""设置全局 access_token"""cls._access_token=token



    @classmethoddefget_access_token(cls):"""获取全局 access_token"""
        returncls._access_tokendef_get_headers(self):

        headers={}ifself.get_access_token():

            headers["Authorization"] = f"Bearer {self.get_access_token()}"
        returnheaders



    asyncdef get(self, url, params=None):

        async with aiohttp.ClientSession() as session:

            async with session.get(

                url, headers=self._get_headers(), params=params

            ) as response:returnawait self._handle_response(response)



    asyncdef post(self, url, json_data=None):

        async with aiohttp.ClientSession() as session:

            async with session.post(

                url, headers=self._get_headers(), json=json_data

            ) as response:returnawait self._handle_response(response)



    asyncdef put(self, url, json_data=None):

        async with aiohttp.ClientSession() as session:

            async with session.put(

                url, headers=self._get_headers(), json=json_data

            ) as response:returnawait self._handle_response(response)



    asyncdef delete(self, url, params=None):

        async with aiohttp.ClientSession() as session:

            async with session.delete(

                url, headers=self._get_headers(), params=params

            ) as response:returnawait self._handle_response(response)



    asyncdef_handle_response(self, response):if response.status == 200:returnawait response.json()else:

            response.raise_for_status()

这些我来基于通用ApiClient的辅助类，对业务接口的调用进行一个简单基类的封装，命名为BaseApi，接受泛型类型定义，如下所示。

classBaseApi(Generic[T]):

    base_url= "http://jsonplaceholder.typicode.com/"client: ApiClient=ApiClient()



    asyncdef getall(self, endpoint, params=None) ->List[T]:

        url= f"{self.base_url}{endpoint}"json_data= await self.client.get(url, params=params)#print(json_data)
        returnlist[T](json_data)



    asyncdef get(self, endpoint, id) ->T:

        url= f"{self.base_url}{endpoint}/{id}"json_data=await self.client.get(url)#return parse_obj_as(T,json_data)
        adapter =TypeAdapter(T)returnadapter.validate_python(json_data)



    asyncdef create(self, endpoint, data) ->bool:

        url= f"{self.base_url}{endpoint}"await self.client.post(url, data)returnTrue



    asyncdef update(self, endpoint, id, data) ->T:

        url= f"{self.base_url}{endpoint}/{id}"json_data=await self.client.put(url, data)



        adapter=TypeAdapter(T)returnadapter.validate_python(json_data)



    asyncdef delete(self, endpoint, id) ->bool:

        url= f"{self.base_url}{endpoint}/{id}"json_data=await self.client.delete(url)#print(json_data)
        return True

我这里使用了一个测试API接口很好的网站：
https://jsonplaceholder.typicode.com/
，它提供了很多不同业务对象的接口信息，如下所示。

统一提供GET/POST/PUT/DELETE等常规Restful动作的处理

如我们获取列表数据的接口如下，返回对应的JSON集合。

通过对应的业务对象不同的动作处理，我们可以测试各种接口。

注意，我们上面的接口都是采用了async/awati的对应异步标识来处理异步的HTTP接口请求。

上面我们定义了BaseApi,具有常规的getall/get/create/update/delete的接口，实际开发的时候，这些会根据后端接口请求扩展更多基类接口。

基于基类BaseApi定义，我们创建其子类PostApi，用来获得具体的对象定义接口。

classPostApi(BaseApi[post]):#该业务接口类，具有基类所有的接口

    #并增加一个自定义的接口
    async def test(self) ->Db:

        url= "http://my-json-server.typicode.com/typicode/demo/db"json_data=await self.client.get(url)#print(json_data)
        return Db.model_validate(json_data)

这里PostApi 具有基类所有的接口:getall/get/create/update/delete的接口, 并可以根据实际情况增加自定义接口，如test 接口定义。

测试代码如下所示。

async defmain():post_api =PostApi()
result= await post_api.getall("posts")print(len(result))



    result= await post_api.get("posts", 1)print(result)



    result=await post_api.create("posts", {"title": "test", "body": "test body", "userId": 1}

    )print(result)



    result=await post_api.update("posts", 1, {"title": "test2", "body": "test body2", "userId": 1, "id": 1}

    )print(result)



    result= await post_api.delete("posts", 1)print(result)



    result=await post_api.test()print(result)if __name__ == "__main__":

    asyncio.run(main())

运行例子，输出如下结果。

使用wxpython开发跨平台桌面应用，对WebAPI调用接口的封装

1、定义Web API接口类并测试API调用基类

2、对异步调用进行测试和接口封装

3、实际HTTTP请求的封装处理

添加新评论

最新文章

最近回复

分类

归档

其它