来自 php 背景,我正在通过 Flask 学习 python。我已经为客户端使用了 WTForms,这很好地处理了验证。
但是,我想使用烧瓶的一件事是公共 API,在这种情况下,我希望所有验证都在我的模型上运行。我认为 SQLAlchemy 将包含验证功能,但事实并非如此。
我遇到过Colander,它看起来不错,但我有点惊讶没有更多无处不在的验证库。更令人惊讶的是 SQLAlchemy 本身并没有这样做。
这里有哪些选择?也许我遗漏了一些东西,但是如何轻松验证模型数据?
来自 php 背景,我正在通过 Flask 学习 python。我已经为客户端使用了 WTForms,这很好地处理了验证。
但是,我想使用烧瓶的一件事是公共 API,在这种情况下,我希望所有验证都在我的模型上运行。我认为 SQLAlchemy 将包含验证功能,但事实并非如此。
我遇到过Colander,它看起来不错,但我有点惊讶没有更多无处不在的验证库。更令人惊讶的是 SQLAlchemy 本身并没有这样做。
这里有哪些选择?也许我遗漏了一些东西,但是如何轻松验证模型数据?
这将允许您拥有一个完美的 DRY 解决方案,因为无论更新源是用户发送的数据,还是作为间接更新的一部分更新模型的应用程序组件,验证都会自动触发。简而言之,您还可以在前端使用 WTForms 重用此解决方案,并且只有一个地方可以对 API 和前端进行验证。
有关在模型中进行验证的更多优点,请参阅此答案。
validates()
装饰器:使用这个装饰器非常简单:只需将它应用于您要验证的字段:
from sqlalchemy.orm import validates
class EmailAddress(Base):
__tablename__ = 'address'
id = Column(Integer, primary_key=True)
email = Column(String)
@validates('email')
def validate_email(self, key, address):
assert '@' in address
return address
当模型实例的属性之一发生更改时,您可以使用属性事件直接执行复杂的验证。使用属性事件的优点是可以保证会话中的数据(内存中的对象)处于验证状态。
这是文档中的一个示例(一个简单的示例,但您应该在这里考虑复杂的规则):
def validate_phone(target, value, oldvalue, initiator):
"Strip non-numeric characters from a phone number"
return re.sub(r'(?![0-9])', '', value)
# setup listener on UserContact.phone attribute, instructing
# it to use the return value
listen(UserContact.phone, 'set', validate_phone, retval=True)
您还可以使用 Mapper Events,例如before_insert
推迟对session.add()
调用的验证,甚至使用Session Events来拦截提交……但是您失去了会话中数据的完整性保证……
我正在为此编写一个库,称为Flask-Inputs。
与滤锅类似,您定义模式并根据它们验证您的输入。就像@Sean Vieira 的建议一样,它依赖于 WTForms 进行验证。
在内部,它将所有request
输入数据转换为 MultiDicts。就像 WTForms 一样,您可以定义自定义验证器(一个内置的自定义验证器用于request.json
数据,它使用 jsonschema 进行验证)。
由于听起来您正在对发布到公共 API 的数据运行验证,因此这里是 API 密钥和发布的 JSON 验证的示例。
from flask_inputs import Inputs
from flask_inputs.validators import JsonSchema
schema = {
'type': 'object',
'properties': {
'name': {'type': 'string'}
}
}
class ApiInputs(Inputs):
headers = {
'Authorization': [DataRequired(), valid_api_key]
}
json = [JsonSchema(schema=schema)]
然后在您的路线中:
@app.route('/api/<version>/endpoint')
def endpoint():
inputs = ApiInputs(request)
if not inputs.validate():
return jsonify(success=False, errors=inputs.errors)
我发现(在生产中使用它)的最大好处是在一个地方显示所有错误。覆盖所有传入数据的良好验证器可以防止生产中出现许多意外/未定义的行为。
只要传入的数据可以以类似 Multi-Dict 的格式读取,就没有理由仍然不能使用 WTForms 进行验证(尽管它比使用滤锅更尴尬)。
因此,对于生成和使用 JSON 的假设 API,您可能会执行以下操作:
class MyDataStructure(Form):
widget = TextField("Widget", validators=[Required()])
quantity = IntegerField("Quantity", validators=[Required()])
@app.route("/api/v1/widgets", methods=["POST"])
def widgets():
try:
new_widget_info = json.loads(request.form.data)
except KeyError:
return jsonify(error="Must provide widget JSON in data param")
except ValueError:
return jsonify(error="Invalid JSON Provided")
data = MyDataStructure(**new_widget_info)
if not data.validate():
return jsonify(error="Missing or invalid data",
error_details=data.errors)
else:
# Create a new widget