2023年8月17日更新,FastAPI比Flask更好用,是快速搭建python api的利器!
Flask学习总结
因为需要调用深度学习模型开发实时接口,模型是基于Python的,所以不可避免的需要整一个Python的Api服务。非实时的接口可以定时调用模型生成结果到数据库,然后Java访问数据库返回结果。
Python Web框架的主流就是Django和Flask,二者各有优劣,Django比较重,学习成本高,考虑下考虑了简单,轻量级的Flask。
用惯了Java的SpringBoot,已经被惯坏了,学习过程中发现Python Web开发远不如Java方便,毕竟Python也不擅长做服务器。从Flask和Django的官方文档也可以发现有一部分内容仍然是前后端不分离的。
本次学习是希望能够将Flask配置成在某些方面像SpringBoot一样开箱即用,主要包含以下几个点。
- 全json交互,统一返回响应体,code,messga,data。
- 统一异常处理,对应Spring全局异常处理。
- 方便快捷的数据校验,对应Spring-validation。
统一响应体
既然是全json交互,主要的问题就在于如何序列化与反序列化,本来以为Python这么灵活的语言,序列化和反序列化的支持化应该很好。本人日常开发中基本只使用到Python标准库json实现序列化,缺点就是默认只能实现基本类型,字典、列表等的序列化反序列化,无法实现自定义对象的序列化与反序列化。
网上查询了很多方案,都没有满足本人灵活、减少硬编码,少配置的要求,也有一些框架的支持,比如django-restframe,marshmallow等,前者需要在django环境下,二者都不太好用,远不如java中的jackson方便快捷。结合实际情况,未使用第三方库。这里贴出个人的解决方案。
Serializable类
class Serializable:
_exclude_fields = ()
def __init__(self, *args, **kwargs):
"""
定义*args,**kwargs参数,接受任何参数
"""
self.__dict__.update(kwargs)
def keys(self) -> typing.Iterable[str]:
"""
返回所有需要序列化的字段
忽略受保护的字段(以_开头)
忽略为None的字段
忽略指定字段,_exclude_fields
"""
return filter(
lambda k: k not in self._exclude_fields and not k.startswith('_') and self.__dict__[k] is not None,
self.__dict__)
def __getitem__(self, key) -> typing.Any:
return getattr(self, key)
可序列化的类继承自Serializable,都有keys方法和__getitem__方法。
在Python中dict方法有如下使用方法
dict() -> new empty dictionary
dict(mapping) -> new dictionary initialized from a mapping object’s
(key, value) pairs
dict(iterable) -> new dictionary initialized as if via:
d = {}
for k, v in iterable:
d[k] = v
dict(**kwargs) -> new dictionary initialized with the name=value pairs
in the keyword argument list. For example: dict(one=1, two=2)
# (copied from class doc)
Serializable有keys方法和__getitem__方法,相当于一个mapping,dict(serializable),首先调用keys()方法返回所有需要序列化的字段,而后对每个字段调用__getitem__方法,这样就可以将类转化为字典了,而Python标准库可以将字典序列化为json字符串。
实体类的定义
"""
dataclass可有可无
"""
@dataclasses.dataclass
class ApiResponse(Serializable):
code: int
message: str
data: typing.Any
@classmethod
def success(cls, data=None):
if data:
return cls(code=200, message='操作成功', data=data)
return cls(code=200, message='操作成功', data=None)
@classmethod
def server_error(cls):
return cls(code=500, message='服务器异常', data=None)
"""
flask-sqlalchemy初始化代码...
"""
class User(db.Model, Serializable):
_exclude_fields = ('password',)
id: Mapped[int] = Column(Integer, primary_key=True, autoincrement=True, nullable=False)
username: Mapped[str] = Column(String(32), nullable=False, server_default='')
password: Mapped[str] = Column(String(64), nullable=False, server_default='123456')
birth_date: Mapped[datetime.date] = Column(Date, nullable=False)
create_time: Mapped[datetime.datetime] = Column(DateTime, nullable=False, server_default=func.now())
def __str__(self):
return f'{self.id}: {self.username}'
__repr__ = __str__
配置json标准库对Serializable类的序列化行为
def config_app(app, test_config=None):
app.config.from_mapping(
SECRET_KEY='dev',
# DATABASE=os.path.join(app.instance_path, 'flaskr.sqlite'),
)
def default(o: typing.Any) -> typing.Any:
"""
自定义default方法
"""
if isinstance(o, Serializable):
return dict(o)
if isinstance(o, datetime.datetime):
return datetime.datetime.strftime(o, "%Y-%m-%d %H:%M:%S")
if isinstance(o, datetime.date):
return datetime.date.strftime(o, '%Y-%m-%d')
if isinstance(o, (decimal.Decimal, uuid.UUID)):
return str(o)
if dataclasses and dataclasses.is_dataclass(o):
return dataclasses.asdict(o)
if hasattr(o, "__html__"):
return str(o.__html__())
raise TypeError(f"Object of type {type(o).__name__} is not JSON serializable")
app.json.default = default # 指定为自定义的default方法
app.json.ensure_ascii = False
app.json.sort_keys = False
app.json.compact = True
if test_config is None:
# load the instance config, if it exists, when not testing
app.config.from_pyfile('config.py')
else:
# load the test config if passed in
app.config.from_mapping(test_config)
flask自带的jsonify函数调用的是DefaultJSONProvider的dumps方法(app.json就是DefaultJSONProvider实例对象)
def dumps(self, obj: t.Any, **kwargs: t.Any) -> str:
"""Serialize data as JSON to a string.
Keyword arguments are passed to :func:`json.dumps`. Sets some
parameter defaults from the :attr:`default`,
:attr:`ensure_ascii`, and :attr:`sort_keys` attributes.
:param obj: The data to serialize.
:param kwargs: Passed to :func:`json.dumps`.
"""
kwargs.setdefault("default", self.default)
kwargs.setdefault("ensure_ascii", self.ensure_ascii)
kwargs.setdefault("sort_keys", self.sort_keys)
return json.dumps(obj, **kwargs)
def _default(o: t.Any) -> t.Any:
if isinstance(o, date):
return http_date(o)
if isinstance(o, (decimal.Decimal, uuid.UUID)):
return str(o)
if dataclasses and dataclasses.is_dataclass(o):
return dataclasses.asdict(o)
if hasattr(o, "__html__"):
return str(o.__html__())
raise TypeError(f"Object of type {type(o).__name__} is not JSON serializable")
标准库的json使用指定的类(cls)参数尝试序列化对象,如果无法序列化,则尝试调用default参数指定的方法,如果仍然无法序列化则报错。可以看到flask默认的_default方法增加了对日期、decimal等类的序列化行为,所以我们只需要重新指定default方法即可。
序列化反序列化测试
class UserView(MethodView):
def get(self):
scalars = db.session.scalars(db.select(User))
return jsonify(ApiResponse.success(scalars.all()))
def post(self):
"""
手动反序列化,且不支持嵌套
"""
user = User(**request.json)
print(user)
print(type(user.username))
return jsonify(ApiResponse.success())
访问接口可查看序列化的json数据。
目前未实现对象的反序列化,只能使用requet.json字典对象。
统一异常处理
只要请求顺利被服务器处理,Http状态码统一为200,统一响应体的code和message字段指示请求正常或错误提示信息,需要对框架抛出的所有的异常和非200状态码进行处理。
Flask全局异常处理的原理和SpringMVC和基本相同的。
def add_error_handler(app: Flask):
@app.errorhandler(Exception)
def handle_all(exp):
current_app.logger.error(exp)
return jsonify(ApiResponse.server_error())
@app.errorhandler(HTTPException)
def handle_http(exp):
current_app.logger.error(exp)
return jsonify(ApiResponse.server_error())
数据字段校验
仿照SpringMVC在解析前端数据,反序列化的时候进行数据校验,使用本人这种简单的序列化方式是很难实现的。
总结
框架内部使用到Python的魔法方法比较多,自己对这方面不太熟悉,功能涉及到底层实现的时候,就比较困难。
Q.E.D.