Location>code7788 >text

Development of cross-platform desktop applications using wxpython, encapsulation of the WebAPI call interface

Popularity:856 ℃/2024-11-13 11:50:56

I introduced in the previous system interface features, including the menu toolbar, business table data, the beginning of the test is based on the simulation of the data, the data is processed in JSON format, through the auxiliary classes to simulate the implementation of the data loading and processing, which is a better way to test the development of the initial period, but the actual business data is certainly from the back-end, including local databases, the SqlServer, Mysql, Oracle, Sqlite, PostgreSQL, etc., or the back-end WebAPI interface to obtain, this accompanying article step by step on how to model the back-end data interface and provide local WebAPI proxy interface class processing.

1, define the Web API interface class and test the API call base class

I wrote about it in the accompanying article.Development of cross-platform desktop applications using wxpython, dynamic tool creation handlingIt introduces the data classes of toolbars and menu bars, and the simulation method to obtain data for display, as shown in the following interface.

For example, the class information of the menu data is shown below.

class MenuInfo:
    id: str  # Menu ID
    pid: str  # Parent Menu ID
    label: str  # Menu Name
    icon: str = None  # menu icon
    path: str = None  # Menu path to locate the view
    tips: str = None  # Menu Tips
    children: list["MenuInfo"] = None

These data are consistent with the definition of the back-end data interface, then it is easy to switch to the dynamic interface.

At the initial stage of system development, we can first try to obtain the data collection in an analog way, such as through a tool to get the data, as shown below.

In order to better match the actual business requirements, we often need to define the information to call the Web API interface based on the server-side interface definition.

We used the Python language in order to develop it all, including the backend, using theBased on SqlAlchemy+Pydantic+FastApi back-end framework

This back-end interface uses a unified interface protocol, the standard protocol is shown below.

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

where result is our data return, which may be of basic type (e.g., string, numeric, boolean, etc.), or a collection of classes, object information, dictionary information, and so on.

If a paged query returns a collection of results, the result is shown below.

Expand the individual record details as shown below.

If we base our definition on the Pydantic model, our Python object class definition code looks like this

from pydantic import  BaseModel
from typing import Generic, Type, TypeVar, Optional
T = TypeVar("T")

# Customized Return Models - Unified Return Results
class AjaxResponse(BaseModel, Generic[T]):
    success: bool = False
    result: Optional[T] = None
    targetUrl: Optional[str] = None
    UnAuthorizedRequest: Optional[bool] = False
    errorInfo: Optional[ErrorInfo] = None

That is, in conjunction with the generic approach, so that the definition can be a good abstraction of different business class interfaces to the base class BaseApi, so that add, delete, change, check and other processing interfaces can be abstracted to the BaseApi inside .

Permission module we involved in the user management, organization management, role management, menu management, function management, operation log, login log and other business classes, then these classes inherit BaseApi, will have the relevant interfaces, the following inheritance relationship.

 

2、Testing and interface encapsulation for asynchronous calls

In order to understand the processing of client-side Api classes, let's first introduce some simple pydantic introductory processing, as follows We start by defining some entity classes to carry data information, as shown below.

from typing import List, TypeVar, Optional, Generic, Dict, Any
from datetime import datetime
from pydantic import BaseModel, Field
T = TypeVar("T")

class AjaxResult(BaseModel, Generic[T]):
    """Testing the Unified Interface Return Format"""
    success: bool = True
    message: Optional[str] = None
    result: Optional[T] = None

class PagedResult(BaseModel, Generic[T]):
    """Pagination results"""
    total: int
    items: List[T]

class Customer(BaseModel):
    """Customer Information Class"""
    name: str
    age: int

The result of the general business is a list of corresponding records, or entity class object format, let's test parsing their JSON data to help us understand.

# Handling of the format of the returned result data
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)

The above parses to the data normally and the output is shown below.

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

If we switch to a uniform return result for testing, it is as follows.

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.)
print(ajax_result.)

The same can be obtained with normal output.

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

The model_validate_json interface allows us to convert the contents of a string to the corresponding business class object, and the model_validate function allows us to convert the JSON format to a business class object.

For the interface inheritance processing, we use the generalized processing, you can greatly reduce the writing of the base class code, such as the following base class definition and subclass definition, you can be a lot simpler, all the logic in the base class processing can be.

class BaseApi(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)
        return result

    def 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)
        return result

    def 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)
        return result


class UserApi(BaseApi[Customer]):
    pass

user_api = UserApi()
result = user_api.getlist()
print()
print()
print()

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

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

As you can see, subclasses only need to specify the inheritance relationship well, without writing any extra code, but then have a specific interface to deal with.

 

3, the actual HTTTP request encapsulation processing

Generally for server-side interface processing, we may need to introduce aiohttp to handle the request, and combined with Pydantic's model processing, is the data can be converted properly, and the same way as the above processing.

First we need to define a generic HTTP request class to handle the return of regular HTTP interface data, as shown below.

class ApiClient:
    _access_token = None  # Class variable for globally shared access_token

    @classmethod
    def set_access_token(cls, token):
        """Setting the global access_token"""
        cls._access_token = token

    @classmethod
    def get_access_token(cls):
        """Get global access_token"""
        return cls._access_token

    def _get_headers(self):
        headers = {}
        if self.get_access_token():
            headers["Authorization"] = f"Bearer {self.get_access_token()}"
        return headers

    async def get(self, url, params=None):
        async with () as session:
            async with (
                url, headers=self._get_headers(), params=params
            ) as response:
                return await self._handle_response(response)

    async def post(self, url, json_data=None):
        async with () as session:
            async with (
                url, headers=self._get_headers(), json=json_data
            ) as response:
                return await self._handle_response(response)

    async def put(self, url, json_data=None):
        async with () as session:
            async with (
                url, headers=self._get_headers(), json=json_data
            ) as response:
                return await self._handle_response(response)

    async def delete(self, url, params=None):
        async with () as session:
            async with (
                url, headers=self._get_headers(), params=params
            ) as response:
                return await self._handle_response(response)

    async def _handle_response(self, response):
        if  == 200:
            return await ()
        else:
            response.raise_for_status()

I've come to base these on the generic ApiClient's helper class, which encapsulates a simple base class for business interface calls, named BaseApi, and accepts generic type definitions, as shown below.

class BaseApi(Generic[T]):
    base_url = "/"
    client: ApiClient = ApiClient()

    async def getall(self, endpoint, params=None) -> List[T]:
        url = f"{self.base_url}{endpoint}"
        json_data = await (url, params=params)
        # print(json_data)
        return list[T](json_data)

    async def get(self, endpoint, id) -> T:
        url = f"{self.base_url}{endpoint}/{id}"
        json_data = await (url)
        # return parse_obj_as(T,json_data)
        adapter = TypeAdapter(T)
        return adapter.validate_python(json_data)

    async def create(self, endpoint, data) -> bool:
        url = f"{self.base_url}{endpoint}"
        await (url, data)

        return True

    async def update(self, endpoint, id, data) -> T:
        url = f"{self.base_url}{endpoint}/{id}"
        json_data = await (url, data)

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

    async def delete(self, endpoint, id) -> bool:
        url = f"{self.base_url}{endpoint}/{id}"
        json_data = await (url)
        # print(json_data)
        return True

I'm using a site here that tests API interfaces very well:/, which provides interface information for many different business objects, as shown below.

Provide unified handling of regular Restful actions such as GET/POST/PUT/DELETE etc.

If we get the list data the interface is as follows, returning the corresponding JSON collection.

We can test various interfaces through different action processing of the corresponding business objects.

Note that our interfaces above all use the async/awati counterpart of the asynchronous identifier to handle asynchronous HTTP interface requests.

Above we defined BaseApi, with the usual getall/get/create/update/delete interfaces, the actual development of these will be based on the back-end interface request to extend more base class interfaces.

Based on the base class BaseApi definition, we create its subclass PostApi, which is used to get the concrete object definition interface.

class PostApi(BaseApi[post]):
    # This business interface class, with all the interfaces of the base class

    # And add a customized interface
    async def test(self) -> Db:
        url = "/typicode/demo/db"
        json_data = await (url)
        # print(json_data)
        return Db.model_validate(json_data)

Here PostApi has all the interfaces of the base class: getall/get/create/update/delete interfaces, and you can add custom interfaces according to the actual situation, such as test interface definition.

The test code is shown below.

async def main():
    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__": (main())

Run the example and output the following result.