标签: webargs

我正在制作一个flask restful api，我遇到的问题是marshmallow-sqlalchemy 和webargs

简而言之，这是我的 sqlalchemy 模型：

class User(Model):
    id = Column(String, primary_key=True)
    name = Column(String(64), nullable=False)
    email = Column(String(120), nullable=False)
    password = Column(String(128))
    creation_date = Column(DateTime, default=datetime.utcnow)

这是我的架构

class UserSchema(ModelSchema):
    class Meta:
        model = User
        strict = True
        sqla_session = db.session

user_schema = UserSchema()

以及使用flask-classful和webargs的路线示例：

class UserView(FlaskView):
    trailing_slash = False
    model = User
    schema = user_schema

    @use_kwargs(schema.fields)
    def post(self, **kwargs):
        try:
            entity = self.model()

            for d in kwargs:
                if kwargs[d] is not missing:
                    entity.__setattr__(d, kwargs[d])

            db.session.add(entity)
            db.session.commit()
            o = …

我将 Flask 与 Flask-restful 和webargs（使用Marshmallow作为后端）一起使用。目前我可以通过以下方式提取我想要的字段：

class AddGroup(Resource):
    args = {
        'name': fields.Str(missing=None),
        'phone': fields.Str(missing=None),
    }

    @use_args(args)
    def get(self, args):
        name = args['name'].strip()
        # ... some GET-related code ...

    @use_args(args)
    def post(self, args):
        name = args['name'].strip()
        # ... some POST-related code ...

到目前为止，一切都很好。但我真正想做的是确保args['name']进入各种方法（“post”、“get”等）时已经删除了空格，这样我就不必每次都手动处理每个变量。（剥离空白只是一个例子 - 它可能是其他一些简单或复杂的转换）

有没有一种方法（通过覆盖 String 字段、定义我自己的字段或其他方式）允许args在到达类方法之前对其进行预处理？

我正在尝试验证发送到我的 api 的所有数据。我的 url 结构中包含一个变量 ，/api/v2/users/<string:username>/collections/以及实际的查询字符串参数，所有这些都需要通过验证。

完整的网址如下所示： https://127.0.0.1:5000/api/v2/users/<string:username>/collections/?page=5&per_page=10

需要验证的两个变量是：username page、 和per_page。简单的解决方案是更改我的 url 结构，但我想知道是否可以完成我需要的内容并保持我当前的简单性，而无需在资源类中添加额外的验证。如果可能的话，你会怎么做？

class UserCollections(Resource):
    @use_args({
        'username': fields.Str(
            required=True,
            validate=username_length,
            error_messages=dict(
                required='Username is required.',
                validator_failed='Username can be between 3 and 25 characters.',
            )
        ),
        'page': fields.Int(
            #required=True,
            missing=1,
            validate=feed_minmax_pages,
            error_messages=dict(
                validator_failed='Maximum number of pages reached.',
            )
        ),
        'per_page': fields.Int(
            #required=True,
            missing=5,
            validate=validate.Range(min=5,max=25),
            error_messages=dict(
                validator_failed='Test number of pages reached.',
            )
        ),
    }, locations=('query',))

    def get(self, args, username):
        print(args)
        print(username)

        return default_schema(
            data={},
            http_status=200
        )

当我运行代码时，我收到验证错误， …