Flask Python模型验证

Jon*_*noB 11 python validation flask

来自php背景,我正在通过Flask学习python.我已经为客户端使用了WTForms,这很好地处理了验证.

但是,我想要使用的东西之一是公共API,在这种情况下,我希望所有验证都在我的模型上运行.我认为SQLAlchemy会包含验证功能,但似乎并非如此.

我遇到了Colander,它看起来很不错,但我有点惊讶的是,没有更多无处不在的验证库.更令人惊讶的是,SQLAlchemy本身并没有这样做.

这里有什么选择?也许我错过了什么,但我怎样才能轻松验证模型数据?

eds*_*ufi 21

您是否考虑过在模型层中进行验证...

这将允许您拥有一个完美的DRY解决方案,因为验证将自动触发,无论更新源是用户发送的数据,还是它是您的应用程序的组件,它正在更新模型作为间接更新的一部分.简而言之,您也可以在WTForms的前端重用此解决方案,并且只有一个地方可以对API和前端进行验证.

有关在模型中进行验证的更多优点,请参阅此答案.


...使用SQLAlchemy提供的工具?

1. 简单验证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
Run Code Online (Sandbox Code Playgroud)

2. 复杂业务规则的ORM事件:

当模型实例的某个属性发生更改时,您可以使用属性事件直接执行复杂验证.使用属性事件的优点是可以保证会话中的数据(内存中的对象)处于验证状态.

这是一个例子(一个简单的例子,但你应该在这里考虑复杂的规则)来自docs:

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)
Run Code Online (Sandbox Code Playgroud)

您还可以使用Mapper事件,例如before_insert将验证推迟到session.add()调用,甚至使用会话事件来拦截提交......但是您会丢失会话中数据的完整性保证...


nat*_*ill 7

我正在为此编写一个名为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)]
Run Code Online (Sandbox Code Playgroud)

然后在您的路线中:

@app.route('/api/<version>/endpoint')
def endpoint():
    inputs = ApiInputs(request)

    if not inputs.validate():
        return jsonify(success=False, errors=inputs.errors)
Run Code Online (Sandbox Code Playgroud)

我发现(在生产中使用它)的最大好处是在一个地方显示所有错误。涵盖所有传入数据的良好验证器可防止生产中出现许多意外/未定义的行为。


Sea*_*ira 2

只要传入的数据可以以类似于 Multi-Dict 的格式读取,就没有理由不能仍然使用 WTForms 进行验证(尽管它比使用 Colander 更尴尬)。

因此,对于生成和使用 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
Run Code Online (Sandbox Code Playgroud)