首页 > 解决方案 > 避免 Flask RESTful 端点中的代码重复

问题描述

我在 Flask RESTful 框架中编写 API。对于每个端点,我都需要检查结构相同但内部流程不同的检查。这是片段。

def get_all_users():
    users = [{"name":"John","dob":"12-10-1990"}, {"name":"David","dob":"25-03-1995"}, {"name":"Maria","dob":"30-02-1998"}]
    return users

端点 1:

class ClassOne(Resource):
    def get(self):
        # Get required info from body
        res_body = request.json
        name = res_body.get('name', None)
        item_list = []
        if name is None:
            users = get_all_users()
            for user in users:
                item_list.append(user['name'])
            return {"nameList":item_list}
        else:
            user = get_user_data(user_info=name)
            if user:
                name = user['name']
                return {"Message":"User {} exists!".format(name)}
            else:
                return {"Error":"User does not exist"}

        return "output based on the above process"

端点 2:

class ClassTwo(Resource):
    def get(self):
        # Get required info from body
        res_body = request.json
        name = res_body.get('name', None)
        item_list = []
        if name is None:
            users = get_all_users()
            for user in users:
                item_list.append(user['dob'])
            return {"nameList":item_list}
        else:
            user = get_user_data(user_info=name)
            if user:
                dob = user['dob']
                return {"Message":"User dob is {}".format(dob)}
            else:
                return {"Error":"User does not exist"}
        
        return "output based on the above process"

通过避免代码重复来编写这些代码的最佳方法是什么。

标签: python-3.xflaskflask-restful

解决方案


假设Resource没有.get(),您的端点类可以继承自一个公共Process类或任何您想调用的类,除了Resource

class Process:
    def get(self, user_key):
        res_body = request.json
        name = res_body.get("name")
        if name is None:
            item_list = []
            users = get_all_users()
            for user in users:
                item_list.append(user.get(user_key))
            return {"nameList": item_list}
        else:
            user = get_user_data(user_info=name)
            if user:
                info = user.get(user_key)
                return {"Message": f"User {user_key} is {info}"}
        return {"Error": "User does not exist"}


class ClassOne(Resource, Process):
    def __init__(self):
        super().__init__()
   
    def get(self):
        return super().get(user_key="name")


class ClassTwo(Resource, Process):
    def __init__(self):
        super().__init__()
   
    def get(self):
        return super().get(user_key="dob")

这是此继承结构的快速演示:

In [1]: class Grandparent:  # Resource
   ...:     pass
   ...:

In [2]: class Parent:  # Process
   ...:     def get(self, x, y):
   ...:         print(x, y)
   ...:

In [3]: class ChildA(Grandparent, Parent):  # Endpoint 1
   ...:     def get(self):
   ...:         super().get(1, 2)
   ...:

In [4]: class ChildB(Grandparent, Parent):  # Endpoint 2
   ...:     def get(self):
   ...:         super().get("hello", "world")
   ...:

In [5]: a = ChildA()

In [6]: b = ChildB()

In [7]: a.get()
1 2

In [8]: b.get()
hello world

或者,您可以简单地Process继承自Resource,然后让您的端点继承自Process.


推荐阅读