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

醉饮千殇别离绪

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

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

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

1. class MenuInfo:
2.     id: str  # 菜单ID
3.     pid: str  # 父菜单ID
4.     label: str  # 菜单名称
5.     icon: str = None  # 菜单图标
6.     path: str = None  # 菜单路径，用来定位视图
7.     tips: str = None  # 菜单提示
8.     children: list["MenuInfo"] = None

复制代码

这些数据和后端数据接口的定义一致，那么就很容易切换到动态的接口上。
在系统开发的初期，我们可以先试用模拟方式获得数据集合，如通过一个工具来来获得数据，如下所示。

为了更好的符合实际的业务需求，我们往往需要根据服务端的接口定义来定义调用Web API接口的信息。
我们为了全部采用Python语言进行开发，包括后端的内容，采用 基于SqlAlchemy+Pydantic+FastApi 的后端框架

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

1. {
2.   "success": false,
3.   <strong>"result":  T ,
4.   </strong>"targetUrl": "string",
5.   "UnAuthorizedRequest": false,
6.   "errorInfo": {
7.     "code": 0,
8.     "message": "string",
9.     "details": "string"
10.   }
11. }

复制代码

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

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

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

1. from pydantic import  BaseModel
2. from typing import Generic, Type, TypeVar, Optional
3. T = TypeVar("T")
5. # 自定义返回模型-统一返回结果
6. class AjaxResponse(BaseModel, Generic[T]):
7.     success: bool = False
8.     result: Optional[T] = None
9.     targetUrl: Optional[str] = None
10.     UnAuthorizedRequest: Optional[bool] = False
11.     errorInfo: Optional[ErrorInfo] = None

复制代码

也就是结合泛型的方式，这样定义可以很好的抽象不同的业务类接口到基类BaseApi中，这样增删改查等处理的接口都可以抽象到BaseApi里面了。
权限模块我们涉及到的用户管理、机构管理、角色管理、菜单管理、功能管理、操作日志、登录日志等业务类，那么这些类继承BaseApi，就会具有相关的接口了，如下所示继承关系。

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

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

1. from typing import List, TypeVar, Optional, Generic, Dict, Any
2. from datetime import datetime
3. from pydantic import BaseModel, Field
4. T = TypeVar("T")
6. class AjaxResult(BaseModel, Generic[T]):
7.     """测试统一接口返回格式"""
8.     success: bool = True
9.     message: Optional[str] = None
10.     result: Optional[T] = None
12. class PagedResult(BaseModel, Generic[T]):
13.     """分页查询结果"""
14.     total: int
15.     items: List[T]
17. class Customer(BaseModel):
18.     """客户信息类"""
19.     name: str
20.     age: int

复制代码

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

1. # 对返回结果数据格式的处理
2. json_data = """{
3.     "total": 100,
4.     "items": [
5.         {"name": "Alice", "age": 25},
6.         {"name": "Bob", "age": 30},
7.         {"name": "Charlie", "age": 35}
8.     ]
9. }"""
10. paged_result = PagedResult.model_validate_json(json_data)
11. print(paged_result.total)
12. print(paged_result.items)

复制代码

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

1. 100
2. [{'name': 'Alice', 'age': 25}, {'name': 'Bob', 'age': 30}, {'name': 'Charlie', 'age': 35}]
3. True

复制代码

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

1. json_data = """{
2.     "success": true,
3.     "message": "success",
4.     "result": {
5.         "total": 100,
6.         "items": [
7.             {"name": "Alice", "age": 25},
8.             {"name": "Bob", "age": 30},
9.             {"name": "Charlie", "age": 35}
10.         ]
11.     }
12. }"""
14. ajax_result = AjaxResult[PagedResult].model_validate_json(json_data)
15. print(ajax_result.success)
16. print(ajax_result.message)
17. print(ajax_result.result.total)
18. print(ajax_result.result.items)

复制代码

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

1. True
2. success
3. 100
4. [{'name': 'Alice', 'age': 25}, {'name': 'Bob', 'age': 30}, {'name': 'Charlie', 'age': 35}]

复制代码

我们通过 model_validate_json 接口可以转换字符串内容为对应的业务类对象，而通过 model_validate 函数可以转换JSON格式为业务类对象。
而对于接口的继承处理，我们采用了泛型的处理，可以极大的减少基类代码的编写，如下基类定义和子类定义，就可以简单很多，所有逻辑放在基类处理即可。

1. class BaseApi(Generic[T]):
2.     def test(self) -> AjaxResult[Dict[str, Any]]:
3.         json_data = """{
4.             "success": true,
5.             "message": "success",
6.             "result": {"name": "Alice", "age": 25}
7.         }"""
8.         result = AjaxResult[Dict[str, Any]].model_validate_json(json_data)
9.         return result
11.     def get(self, id: int) -> AjaxResult[T]:
12.         json_data = """{
13.             "success": true,
14.             "message": "success",
15.             "result": {"name": "Alice", "age": 25}
16.         }"""
17.         result = AjaxResult[T].model_validate_json(json_data)
18.         return result
20.     def getlist(self) -> AjaxResult[List[T]]:
21.         json_data = """{
22.             "success": true,
23.             "message": "success",
24.             "result": [
25.                 {"name": "Alice", "age": 25},
26.                 {"name": "Bob", "age": 30},
27.                 {"name": "Charlie", "age": 35}
28.             ]
29.         }"""
30.         result = AjaxResult[List[T]].model_validate_json(json_data)
31.         return result
34. <strong>class UserApi(BaseApi[Customer]):
35.     </strong>pass
37. user_api = UserApi()
38. result = user_api.getlist()
39. print(result.success)
40. print(result.message)
41. print(result.result)
43. result = user_api.get(1)
44. print(result.success)
45. print(result.message)
46. print(result.result)
48. result = user_api.test()
49. print(result.success)
50. print(result.message)
51. print(result.result)

复制代码

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

一般对于服务端接口的处理，我们可能需要引入 aiohttp 来处理请求，并结合Pydantic的模型处理，是的数据能够正常的转换，和上面的处理方式一样。
首先我们需要定义一个通用HTTP请求的类来处理常规的HTTP接口数据的返回，如下所示。

1. class ApiClient:
2.     _access_token = None  # 类变量，用于全局共享 access_token
4.     @classmethod
5.     def set_access_token(cls, token):
6.         """设置全局 access_token"""
7.         cls._access_token = token
9.     @classmethod
10.     def get_access_token(cls):
11.         """获取全局 access_token"""
12.         return cls._access_token
14.     def _get_headers(self):
15.         headers = {}
16.         if self.get_access_token():
17.             headers["Authorization"] = f"Bearer {self.get_access_token()}"
18.         return headers
20.     async def get(self, url, params=None):
21.         async with aiohttp.ClientSession() as session:
22.             async with session.get(
23.                 url, headers=self._get_headers(), params=params
24.             ) as response:
25.                 return await self._handle_response(response)
27.     async def post(self, url, json_data=None):
28.         async with aiohttp.ClientSession() as session:
29.             async with session.post(
30.                 url, headers=self._get_headers(), json=json_data
31.             ) as response:
32.                 return await self._handle_response(response)
34.     async def put(self, url, json_data=None):
35.         async with aiohttp.ClientSession() as session:
36.             async with session.put(
37.                 url, headers=self._get_headers(), json=json_data
38.             ) as response:
39.                 return await self._handle_response(response)
41.     async def delete(self, url, params=None):
42.         async with aiohttp.ClientSession() as session:
43.             async with session.delete(
44.                 url, headers=self._get_headers(), params=params
45.             ) as response:
46.                 return await self._handle_response(response)
48.     async def _handle_response(self, response):
49.         if response.status == 200:
50.             return await response.json()
51.         else:
52.             response.raise_for_status()

复制代码

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

1. class BaseApi(Generic[T]):
2.     <strong>base_url </strong><strong>= "http://jsonplaceholder.typicode.com/"</strong>
3.     client: ApiClient = ApiClient()
5.     async def getall(self, endpoint, params=None) -> List[T]:
6.         url = f"{self.base_url}{endpoint}"
7.         json_data = await self.client.get(url, params=params)
8.         # print(json_data)
9.         return list[T](json_data)
11.     async def get(self, endpoint, id) -> T:
12.         url = f"{self.base_url}{endpoint}/{id}"
13.         json_data = await self.client.get(url)
14.         # return parse_obj_as(T,json_data)
15.         adapter = TypeAdapter(T)
16.         return adapter.validate_python(json_data)
18.     async def create(self, endpoint, data) -> bool:
19.         url = f"{self.base_url}{endpoint}"
20.         await self.client.post(url, data)
22.         return True
24.     async def update(self, endpoint, id, data) -> T:
25.         url = f"{self.base_url}{endpoint}/{id}"
26.         json_data = await self.client.put(url, data)
28.         adapter = TypeAdapter(T)
29.         return adapter.validate_python(json_data)
31.     async def delete(self, endpoint, id) -> bool:
32.         url = f"{self.base_url}{endpoint}/{id}"
33.         json_data = await self.client.delete(url)
34.         # print(json_data)
35.         return True

复制代码

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

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

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

 通过对应的业务对象不同的动作处理，我们可以测试各种接口。 
 注意，我们上面的接口都是采用了async/awati的对应异步标识来处理异步的HTTP接口请求。
上面我们定义了BaseApi,具有常规的getall/get/create/update/delete的接口，实际开发的时候，这些会根据后端接口请求扩展更多基类接口。
基于基类BaseApi定义，我们创建其子类PostApi，用来获得具体的对象定义接口。

1. class PostApi(BaseApi[post]):
2.     # 该业务接口类，具有基类所有的接口
4.     # 并增加一个自定义的接口
5.     async def test(self) -> Db:
6.         url = "http://my-json-server.typicode.com/typicode/demo/db"
7.         json_data = await self.client.get(url)
8.         # print(json_data)
9.         return Db.model_validate(json_data)

复制代码

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

1. async def main():<strong>
2.     post_api =</strong><strong> PostApi()</strong><br>
3.     result = await post_api.getall("posts")
4.     print(len(result))
6.     result = await post_api.get("posts", 1)
7.     print(result)
9.     result = await post_api.create(
10.         "posts", {"title": "test", "body": "test body", "userId": 1}
11.     )
12.     print(result)
14.     result = await post_api.update(
15.         "posts", 1, {"title": "test2", "body": "test body2", "userId": 1, "id": 1}
16.     )
17.     print(result)
19.     result = await post_api.delete("posts", 1)
20.     print(result)
22.     result = await post_api.test()
23.     print(result)
26. if __name__ == "__main__":
27.     asyncio.run(main())

复制代码

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

 

来源:https://www.cnblogs.com/wuhuacong/p/18543247
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！