前记 给接口加上类型检查的想法源于我实习时写接口的灵感, 那时候就是一个crud boy,一直在写接口,但每次都要在接口写一些参数校验或参数类型转换, 觉得很繁琐,总觉得这个是可以自己封装一个,减少大量代码. 在实现的过程中发现自己慢慢的接受了代码即文档的实现, 并且慢慢迭代, 使之越来越完善, 最终蜕变成一个开源项目pait
注 : 第一版是18年实习的时候,基 于一个叫sanic的web框架以及Python3.5写的.
19/03 更新: 为了代码统一,这里的示例是把sanic的接口代码迁移到starlette, Python3.7了.
19/04 更新: 增加Pydantic的使用
20/07 更新: 根据几个版本迭代出了项目pait ,功能更多,并且支持接口文档导出, 以及支持swagger和redoc页面展示.
21/01 更新: 更新每个版本的完整代码跳转链接
1.为何要加上给接口加上一层类型校验 1.1.示例接口 先看看一个普通的示例接口:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 import uvicornfrom starlette.applications import Starlettefrom starlette.responses import JSONResponsefrom starlette.routing import Route, Routerasync def demo_post (request ): json_data = await request.json() uid = request_data.get('uid' , 0 ) timestamp = request_data.get('timestamp' , 0 ) user_info = request_data.get('user_info' , {}) user_name = request_data.get('user_name' , '' ) return JSONResponse({'hello' : 'world' }) app = Starlette( routes=[ Route('/api' , demo_post, methods=['POST' ]), ] ) uvicorn.run(app)
可以看到, 这个接口接收POST请求, 路由函数获取了4个参数uid
,timestamp
,user_info
,user_name
.在正常的接口中需要对参数进行处理和转换,比如uid这里只接受int类型,那么就需要多写一个判断或者转换, 如下:
1 2 3 4 5 6 7 try uid = int (uid)except Exception as e: raise eif type (uid) != int : raise Exception
可以看到平时写这种参数校验或类型转换会比较繁琐,同时接口的所有参数都需要参数校验,非常麻烦.一般像这些又繁琐又简单的东西都可以给程序自动化处理的,减轻自己的负担.
2.接口类型校验加在哪里? 从上面的例子来看,加上一层接口类型自动校验和转换是非常有用且利于我们写接口的,也能减少很多工作量,那么这一层得怎么加,加在哪呢?
这一层主要工作是拿到请求信息, 并进行转换和处理, 再把结果传递给路由函数, 那么, 我们这一层必须处于在框架拿到请求信息之后, 在进入路由函数之前. 而在框架中,只有中间件和给路由函数套上装饰器才可以比较优雅的增加接口类型校验. 但考虑到每个接口的参数都不一致,所以只能把接口类型校验这一层封装在一个装饰器里面,并在对应的路由函数套上该装饰器.
既然决定了只能把接口类型校验放在装饰器,那只能通过传递参数和编写装饰器来共同完成接口类型校验功能,其中装饰器的逻辑是绝对不会变的, 应传递的参数决定了该路由函数接受了什么样的参数,那最好就是传递的参数是一个model(也就是python的类型),如果有新加功能,就是给类添加一个方法,再由装饰器的逻辑去调用.
好了思路已经想好了,开始实现吧.
2.1 第一版,使用python基本类型校验 第一版代码
第一版是整个代码的核心部分,之后也是围绕第一版代码进行修改,这里先贴所有代码 相对于示例接口, 主要改动是:
1.增加一个Model类,用户通过继承可以简单快速的创建自己想要的Mode
2.增加一个装饰器,用户可以把要检验的函数套上装饰器,并传入自己定义的Model.
3.增加一个``request.state.param_dict`的方法, 路由函数可以直接从该方法获得已经被转换好的数据
其他说明见代码里面的注释:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 import astfrom functools import wrapsfrom typing import Callable, Typeimport uvicornfrom starlette.applications import Starlettefrom starlette.requests import Requestfrom starlette.responses import JSONResponsefrom starlette.routing import Routeclass Model : """创建一个Model类, 类属性是{参数}={参数类型}""" def __init__ (self ): """这里把类属性的值设置到__dict__""" for key in self.__dir__(): if key.startswith('_' ): continue value = getattr (self, key, None ) if not isinstance (value, type ): continue self.__dict__[key] = value def to_dict (self ): """把{参数}={类型}转为dict""" return self.__dict__class CustomModel (Model ): uid = int timestamp = int user_info = dict user_name = str def params_verify (model: Type[Model] ): """装饰器""" def wrapper (func: Callable ): @wraps(func ) async def request_param (request: Request, *args, **kwargs ): param_dict: dict = dict (request.query_params) if request.method == "POST" : param_dict.update(await request.json()) instance_model: Model = model() try : for key, key_type in instance_model.to_dict().items(): if key in param_dict: value = ast.literal_eval(param_dict[key]) param_dict[key] = key_type(value) request.state.param_dict = param_dict return await func(request, *args, **kwargs) except Exception as e: return JSONResponse({'error' : str (e)}) return request_param return wrapper@params_verify(CustomModel ) async def demo_post (request ): return JSONResponse({'result' : request.state.param_dict})@params_verify(CustomModel ) async def demo_get (request ): return JSONResponse({'result' : request.state.param_dict}) app = Starlette( routes=[ Route('/api' , demo_post, methods=['POST' ]), Route('/api' , demo_get, methods=['GET' ]), ] )if __name__ == "__main__" : uvicorn.run(app)
代码编写好后开始运行代码,测试GET:/api
1 2 3 4 >>> curl "127.0.0.1:8000/api?uid=123" {"result" :{"uid" :123}} >>> curl "127.0.0.1:8000/api?uid=abc" {"error" :"malformed node or string: <_ast.Name object at 0x7f3b2c5c3c10>" }
可以看到当uid=123时,程序会自动把123转换为int类型, 而uid=abc则无法进行转换,并由ast模块抛错. 看得出第一版还是挺不错的,但是使用上有些麻烦,接下来进入第二版,添加更多常用的功能.
2.2 第二版,添加其他常用的功能 第二版代码
在做接口检验时, 除了做类型识别外, 我们还会有一些常见的需求:
字符串长度的最大/小长值
数字最大/最小值
该参数是否是必填
但是由于我们设计Model时,采用’{参数}={类型}’的方法,已经填不了更多的值了,如果传一个dict或者list 限制又比较多,同时会让人很迷惑,同时,装饰器函数只是个函数,功能尽量的单一,要方便拓展.所以在第二版,我结合Python的描述器协议,对Model进行改版.
首先创建一个MISS_OBJECT
空对象, 用来判断用户有没有设置默认值, 和CustomValueError
异常
1 2 3 4 5 6 class CustomValueError (ValueError ): """异常""" pass MISS_OBJECT: 'object()' = object ()
接下来就是一个创建Field,我们可以通过Field来实现拓展功能,Field是一个Python描述器,用于托管Model类属性的__get__
,__set__
,__delete__
方法,本次修改的主要功能都在Field
的__set__
方法里面(目前简单就都写在同一个函数里)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 class Field : def __init__ ( self, _type: Type, default: Any = MISS_OBJECT, max_length: Optional[int ] = None , max_value: Optional[int ] = None , min_value: Optional[int ] = None , ): self._type: Type = _type self._default: Any = default self._max_length: Optional[int ] = max_length self._max: Optional[int ] = max_value self._min: Optional[int ] = min_value self._dict: Dict[str , Any] = {} def __get__ (self, instance: 'object()' , owner: Type ) -> Any: try : value = self._dict[instance] except Exception as e: raise CustomValueError('value must not empty' ) from e return value def __set__ (self, instance: 'object()' , value: Union[str , Any] ): if value is MISS_OBJECT: if self._default is not MISS_OBJECT: value = self._default else : raise ValueError('value must not empty' ) if value is not None : if type (value) != self._type: value = ast.literal_eval(value) value = self._type(value) if isinstance (value, str ) or isinstance (value, list ) or isinstance (value, tuple ): if self._max_length and len (value) > self._max_length: raise ValueError(f'value length:{len (value)} > {self._max_length} ' ) elif isinstance (value, int ) or isinstance (value, float ): if self._max is not None and value > self._max: value = self._max elif self._min is not None and value < self._min: value = self._min self._dict[instance] = value def __delete__ (self, instance ): del self._dict[instance]
编写完了描述器,那么我们可以根据描述器来更改Model了, __init__
的初始化方法进行更改,把用到Field的属性加入到field_list
中,方便参数校验装饰器调用.在CustomModel
中我们赋值Field,并使用FIeld中的功能约束属性.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 class Model : def __init__ (self ): """这里把类属性的值设置到field_dict""" self.field_list: List[str ] = [] for key in self.__dir__(): if key.startswith('_' ): continue if key in self.__dict__: continue try : getattr (self, key) except CustomValueError: self.field_list.append(key) def to_dict (self ) -> Dict[str, Any]: """把{参数}={类型}转为dict""" return { key: getattr (self, key) for key in self.field_list }class CustomModel (Model ): uid = Field(int , min_value=10 , max_value=100 ) timestamp = Field(int , default=None ) user_info = Field(dict , default=None ) user_name = Field(str , max_length=4 )
Model改造好了,可以把原来的装饰器进行修改:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 def params_verify (model: Type[Model] ): """装饰器""" def wrapper (func: Callable ): @wraps(func ) async def request_param (request: Request, *args, **kwargs ): param_dict: dict = dict (request.query_params) if request.method == "POST" : param_dict.update(await request.json()) instance_model: Model = model() try : for key in instance_model.field_list: value = param_dict.get(key, MISS_OBJECT) setattr (instance_model, key, value) request.state.model = instance_model return await func(request, *args, **kwargs) except Exception as e: return JSONResponse({'error' : str (e)}) return request_param return wrapper@params_verify(CustomModel ) async def demo_post (request ): return JSONResponse({'result' : request.state.model.to_dict()})@params_verify(CustomModel ) async def demo_get (request ): return JSONResponse({'result' : request.state.model.to_dict()})
改写完了,继续跑下测试,看看结果怎么样(由于还没完善异常处理,所以只抛出简单的异常信息)
1 2 3 4 5 6 7 8 9 >>> curl "127.0.0.1:8000/api?uid=123" {"error" :"value must not empty" } >>> curl "127.0.0.1:8000/api?uid=123&user_name=apple" {"error" :"value length:5 > 4" } >>> curl "127.0.0.1:8000/api?uid=123&user_name=appl" {"result" :{"uid" :100,"timestamp" :null,"user_info" :null,"user_name" :"appl" }}
可以看到这一版本还有一些细节还没敲定好, 但是基本功能都是完备的, 也可以很方便的进行校验功能拓展.
2.3 第三版,使用Type Hints 第三版代码
在Python 3.5之后开始出现了一个叫TypeHints的东西, TypeHints可以通过类似于
1 2 def test (a: int , b: str ) -> list return [a, b]
来告诉 IDE或者代码检查工具,查看用户写代码的类型是否准确(但是会让你感觉在写静态语言- -), 但不会影响运行时的性能. 在3.5时, TypeHints非常简单, 如今到了3.7TypeHint已经越来越完善了, 我们可以把Field的type转移到annotation中, 这样在写路由函数时, IDE可以帮我们更好的分析, 防止我们写错代码. 这一版本中, 由于使用了Python的TypeHint, 所以Field不用传入type了, 把类型写在annotation中.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 class Field : ... def type_handle (self, value: Any, key_type: Union[Type, _GenericAlias] ) -> Any: """兼容TyepHint和python基础类型转换的handle 目前只支持typing的Union,Option和所有Python的基础类型 """ if hasattr (key_type, '__origin__' ) and key_type.__origin__ is Union: key_type = key_type.__args__ if not isinstance (value, key_type): try : if isinstance (key_type, tuple ): for i in key_type: try : value = self._python_type_conversion(i, value) break except TypeError: value = None else : raise TypeError else : value = self._python_type_conversion(key_type, value) except Exception: raise TypeError(f"The type should be {key_type} " ) return value @staticmethod def _python_type_conversion (key_type: Type, value: str ) -> Any: """Python基础类型转换""" value = ast.literal_eval(value) if type (value) == key_type: return value try : return key_type(value) except Exception: raise TypeError(f"Value type:{type (value)} is not {key_type} " ) def __set__ (self, instance: 'object()' , value_tuple: Tuple[Any, Union[Type, _GenericAlias]] ): value, key_type = value_tuple if value is MISS_OBJECT: if self._default is not MISS_OBJECT: value = self._default else : raise ValueError('value must not empty' ) if value is not None : value = self.type_handle(value, key_type) if isinstance (value, str ) or isinstance (value, list ) or isinstance (value, tuple ): if self._max_length and len (value) > self._max_length: raise ValueError(f'value length:{len (value)} > {self._max_length} ' ) elif isinstance (value, int ) or isinstance (value, float ): if self._max is not None and value > self._max: value = self._max elif self._min is not None and value < self._min: value = self._min self._dict[instance] = value
使用TypeHint后,类型信息存放点从Field改为Model,所以Model也要做适配性改造
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 class Model : def __init__ (self ): """这里把类属性的值设置到field_dict""" self.field_dict = {} for key in self.__annotations__: if key.startswith('_' ): continue if key in self.__dict__: continue try : getattr (self, key) except CustomValueError: self.field_dict[key] = self.__annotations__[key] def to_dict (self ): """把{参数}={类型}转为dict""" return { key: getattr (self, key) for key in self.field_dict }
装饰器函数同样针对Model和Field的改动进行适配
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 def params_verify (model: Type[Model] ): """装饰器""" def wrapper (func: Callable ): @wraps(func ) async def request_param (request: Request, *args, **kwargs ): param_dict: dict = dict (request.query_params) if request.method == "POST" : param_dict.update(await request.json()) instance_model: Model = model() try : for key, key_type in instance_model.field_dict.items(): value = param_dict.get(key, MISS_OBJECT) setattr (instance_model, key, (value, key_type)) request.state.model = instance_model return await func(request, *args, **kwargs) except Exception as e: return JSONResponse({'error' : str (e)}) return request_param return wrapper
进行修改后,CustomModel就可以支持TyepHint功能了, 可以重点关注uid的类型,现在可以同时支持str和int了.
1 2 3 4 5 class CustomModel (Model ): uid: Union[str , int ] = Field(min_value=10 , max_value=100 ) timestamp: int = Field(default=None ) user_info: dict = Field(default=None ) user_name: str = Field(max_length=4 )
接下来看以下测试,,由于Field的逻辑,第一个类型检查转换完毕后,就不会继续检查和转换.所以uid会优先被转换为str,并不受min和max的影响.可以看到下面返回的uid的值是“123”
,并不会受到max的限制
1 2 >>> curl "127.0.0.1:8000/api?uid=123&user_name=appl" {"result" :{"uid" :"123" ,"timestamp" :null,"user_info" :null,"user_name" :"appl" }}
2.4 第四版,使用inspect 第四版代码
在TypeHint出现后,inspect可以解析函数的参数以及参数对应的值,同时也能解析我们标注函数返回的类型,这样我们就可以做2个改造了:
1.把model从装饰器传入改为从函数传入,从函数传入有多个优点:
调用者简单明了,知道这就是他自己想要的参数
支持多个model,用户可以重复利用多个model.
支持param: type = value的形式(这里没实现,可以见项目pait )
2.对返回类型进行标注,同时进行检查和抛错
一般来说第一点比较重要,可能有人觉得第二点导致他无法返回其他状态码响应,解决办法是从装饰器转入状态码即可.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 class CustomModel (Model ): uid: Union[str , int ] = Field(min =10 , max =100 ) timestamp: int = Field(default=None ) user_info: dict = Field(default=None ) user_name: str = Field(max_length=4 )class CustomOtherModel (Model ): age: int = Field(min =1 , max =100 )def params_verify (): """装饰器""" def wrapper (func: Callable ): @wraps(func ) async def request_param (request: Request, *args, **kwargs ): param_dict: dict = dict (request.query_params) if request.method == "POST" : param_dict.update(await request.json()) sig: 'inspect.Signature' = inspect.signature(func) fun_param_dict: Dict[str , inspect.Parameter] = { key: sig.parameters[key] for key in sig.parameters if sig.parameters[key].annotation != inspect._empty } return_param: Type = sig.return_annotation try : func_args = [] for param in fun_param_dict.values(): model: Model = param.annotation() for key, key_type in model.field_dict.items(): value: Any = param_dict.get(key, MISS_OBJECT) setattr (model, key, (value, key_type)) func_args.append(model) response = await func(request, *func_args, **kwargs) if type (response) != return_param: raise ValueError(f'response type != {return_param} ' ) if dict is return_param: return JSONResponse(response) raise TypeError(f'{type (response)} not support' ) except Exception as e: return JSONResponse({'error' : str (e)}) return request_param return wrapper@params_verify() async def demo_get (request, model: CustomModel, other_model: CustomOtherModel ) -> dict: return_dict = model.to_dict() return_dict.update(other_model.to_dict()) return {'result' : return_dict}
修改完成后,进行测试,可以发现age被转为min=1的值.
1 2 >>>curl "127.0.0.1:8000/api?uid=123&user_name=appl&age=-1" {"result" :{"uid" :"123" ,"timestamp" :null,"user_info" :null,"user_name" :"appl" ,"age" :1}}
2.5 第五版,使用Pydantic 第五版代码
其实到了第四版,就开始投入使用,并运行很长的时间了,因为对类型嵌套没有比较大需求,所以也没打算有什么改进.
直到有一天看到fastapi同时又发现自己其实是有类型嵌套的需求, 所以就开始了解Pydantic并使用他(毕竟功能类似,功能还能更多,何乐而不为呢).
Pydantic功能非常多,远远大于上面的Model和Field, 支持多种校验, 支持Orm, 支持openapi schema生成等等,需要什么功能,从文档 里查找就可以了.
由于之前实现Model的逻辑跟Pydantic的BaseModel逻辑很像, 所以基于第四版代码修改并使用Pydantic十分简单,只要把自己的Model,Field删除掉,并且引入Pydantic即可(如下),并在装饰器那里使用dict的方式传入并把调用model的to_dict改为dict即可
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 import inspectfrom functools import wrapsfrom typing import Any, Dict, List, Typeimport uvicornfrom starlette.applications import Starlettefrom starlette.requests import Requestfrom starlette.responses import JSONResponsefrom starlette.routing import Routefrom pydantic import ( BaseModel, conint, constr, )class PydanticModel (BaseModel ): uid: conint(gt=10 , lt=1000 ) user_name: constr(min_length=2 , max_length=4 )class PydanticOtherModel (BaseModel ): age: conint(gt=1 , lt=100 )def params_verify (): """装饰器""" def wrapper (func ): @wraps(func ) async def request_param (request: Request, *args, **kwargs ): param_dict: dict = dict (request.query_params) if request.method == "POST" : param_dict.update(await request.json()) sig: 'inspect.Signature' = inspect.signature(func) fun_param_dict: Dict[str , inspect.Parameter] = { key: sig.parameters[key] for key in sig.parameters if sig.parameters[key].annotation != inspect._empty } return_param: Any = sig.return_annotation try : func_args: List[BaseModel] = [] for param in fun_param_dict.values(): if param.annotation is Request: continue model: BaseModel = param.annotation(**param_dict) func_args.append(model) response: Any = await func(request, *func_args, **kwargs) if type (response) != return_param: raise ValueError(f'response type != {return_param} ' ) if dict is return_param: return JSONResponse(response) raise TypeError(f'{type (response)} not support' ) except Exception as e: return JSONResponse({'error' : str (e)}) return request_param return wrapper@params_verify() async def demo_get (request: Request, model: PydanticModel, other_model: PydanticOtherModel ) -> dict: return_dict = model.dict () return_dict.update(other_model.dict ()) return {'result' : return_dict} app = Starlette( routes=[ Route('/api' , demo_get, methods=['GET' ]), ] )if __name__ == "__main__" : uvicorn.run(app)
继续跑刚才的测试,由于逻辑有点不同,Pydantic不会进行自动转换,但是抛错信息挺详细的(改为e.json报错会更详细)…
1 2 3 4 >>> curl "127.0.0.1:8000/api?uid=123&user_name=apple&age=-1" {"error" :"1 validation error for PydanticModel\nuser_name\n ensure this value has at most 4 characters (type=value_error.any_str.max_length; limit_value=4)" } >>> curl "127.0.0.1:8000/api?uid=123&user_name=appl&age=2" {"result" :{"uid" :123,"user_name" :"appl" ,"age" :2}}
到这里,对于接口层的类型校验基本功能的升级基本完成了,可以看到理解了原理后是非常简单的代码(在引入Pydantic后),但减少了很多重复的代码量,再稍微拓展,就可以支持更多的框架啦.
2.6 更多功能 这份代码很简单, 主要做了对接口的参数检查,转换等封装, 但是只获取了url的参数和json body参数, 在日常使用中我还更希望我能快速的获取到cookie和header的参数. 同时由于代码既文档的思想, 我还想让他能自动生成文档, 所以我再进行了升级改造, 使他能实现这些功能,详情查看我的开源项目pait , 他还支持cbv模式, 支持更好的错误报错, 更重要的是支持更多的框架.
3.总结 在第一次编写接口的类型校验时, 带我的大佬说不提倡这样写, 特别是不要经常引入装饰器(此时处于python2与python3的转换期),觉得装饰器会让人误解,导致别人很难一下子看得懂逻辑, 在路由函数的doc写就可以了, 但是后面慢慢多了TypeHint, 也有Pydantic这样优秀的框架, 我觉得实现这个类型校验带来的收益(比如减少代码量等)远远大于他让人难以理解时,那我们就可以选择使用他.所以我就从实现后一直使用这个功能,并且逐渐的完善他,让他更好用,同时也让我对Python的TypeHint更加的深入理解和运用,现在写非小脚本的Python代码时也都在用TypeHint了,推荐大家都使用他.随着TypeHint的发展,在大项目中写Python会越来越顺手的, 也能让Python项目更工程化.