天天看點

Flask 學習-48.Flask-RESTX 使用api.model() 模型工廠前言api.model() 工廠使用示例@api.expect 裝飾器啟用或禁用 validate 驗證:@api.marshal_with()裝飾器使用示例

前言

model()工廠允許您将模型執行個體化并注冊到您的API或Namespace.

api.model() 工廠

有2種使用方式,第一種直接使用 api.model

my_fields = api.model('MyModel', {
    'name': fields.String,
    'age': fields.Integer(min=0)
})           

複制

第二種間接注冊到api,以下方式是等價的

# Equivalent to
my_fields = Model('MyModel', {
    'name': fields.String,
    'age': fields.Integer(min=0)
})
api.models[my_fields.name] = my_fields           

複制

使用示例

user模型

class Users(db.Model):
    __tablename__ = 'user'  # 資料庫表名
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(50), unique=True, nullable=False)
    password = db.Column(db.String(128), nullable=False)
    is_active = db.Column(db.Boolean, default=1)
    email = db.Column(db.String(64), nullable=True)
    create_time = db.Column(db.DateTime, default=datetime.now)
    update_time = db.Column(db.DateTime, onupdate=datetime.now, default=datetime.now)

    def hash_password(self, password):
        """密碼加密"""
        self.password = sha256_crypt.encrypt(password)

    def verify_password(self, password):
        """校驗密碼"""
        return sha256_crypt.verify(password, self.password)

    def __repr__(self):
        return f"<Users(id='{self.id}', username='{self.username}'...)>"           

複制

校驗請求入參,除了之前學到的 RequestParser 來定義預期的輸入:

@api.route('/api/user', endpoint='user')
class UserView(Resource):

    def post(self):
        """add user"""
        parser = reqparse.RequestParser()
        parser.add_argument('username', required=True, type=str, help='username is required')
        parser.add_argument('password', required=True, type=str, help='password is required')
        args = parser.parse_args()
        print(f'請求參數: {args}')
        # 儲存資料庫
        return {"msg": "create success"}           

複制

也可以使用api.model() 的方法來校驗請求入參, 以下方式是等價的

from flask_restx import Resource, fields

user_input = api.model('UserModel', {
    'username': fields.String(required=True),
    'password': fields.String(required=True)
})

@api.route('/api/user', endpoint='user')
class UserView(Resource):

    @api.expect(user_input, validate=True)
    def post(self):
        """add user"""
        print(f'請求參數: {api.payload}')
        # 儲存資料庫
        return {"msg": "create success"}           

複制

@api.expect 裝飾器

裝飾器

@api.expect()

允許您指定預期的輸入字段。它接受一個可選的布爾參數

validate

,訓示是否應驗證有效payload 參數。

RESTX_VALIDATE可以通過将配置設定為True 或傳遞validate=True給 API 構造函數來全局自定義驗證行為。

以下示例是等效的, 未設定

validate=True

功能等價于

@api.expect()

功能跟

api.doc()

一樣(

api.doc()

用于 swagger 文檔輸出)

使用@api.expect()裝飾器:

resource_fields = api.model('Resource', {
    'name': fields.String,
})

@api.route('/my-resource/<id>')
class MyResource(Resource):
    @api.expect(resource_fields)
    def get(self):
        pass           

複制

使用api.doc()裝飾器:

resource_fields = api.model('Resource', {
    'name': fields.String,
})

@api.route('/my-resource/<id>')
class MyResource(Resource):
    @api.doc(body=resource_fields)
    def get(self):
        pass           

複制

您可以将清單指定為預期輸入:

resource_fields = api.model('Resource', {
    'name': fields.String,
})

@api.route('/my-resource/<id>')
class MyResource(Resource):
    @api.expect([resource_fields])
    def get(self):
        pass           

複制

您可以使用RequestParser來定義預期的輸入:

parser = api.parser()
parser.add_argument('param', type=int, help='Some param', location='form')
parser.add_argument('in_files', type=FileStorage, location='files')

@api.route('/with-parser/', endpoint='with-parser')
class WithParserResource(restx.Resource):
    @api.expect(parser)
    def get(self):
        return {}           

複制

啟用或禁用 validate 驗證:

可以在特定端點上啟用或禁用驗證:

resource_fields = api.model('Resource', {
    'name': fields.String,
})

@api.route('/my-resource/<id>')
class MyResource(Resource):
    # Payload validation disabled  禁用validate 校驗入參
    @api.expect(resource_fields)
    def post(self):
        pass

    # Payload validation enabled  啟用validate 校驗入參
    @api.expect(resource_fields, validate=True)
    def post(self):
        pass           

複制

通過配置進行應用程式範圍驗證的示例:

app.config['RESTX_VALIDATE'] = True

api = Api(app)

resource_fields = api.model('Resource', {
    'name': fields.String,
})

@api.route('/my-resource/<id>')
class MyResource(Resource):
    # Payload validation enabled
    @api.expect(resource_fields)
    def post(self):
        pass

    # Payload validation disabled
    @api.expect(resource_fields, validate=False)
    def post(self):
        pass           

複制

通過構造函數進行應用程式範圍驗證的示例:

api = Api(app, validate=True)

resource_fields = api.model('Resource', {
    'name': fields.String,
})

@api.route('/my-resource/<id>')
class MyResource(Resource):
    # Payload validation enabled
    @api.expect(resource_fields)
    def post(self):
        pass

    # Payload validation disabled
    @api.expect(resource_fields, validate=False)
    def post(self):
        pass           

複制

@api.marshal_with()裝飾器

這個裝飾器像原始marshal_with()裝飾器一樣工作,不同之處在于它記錄了方法。可選參數code允許您指定預期的 HTTP 狀态代碼(預設為 200)。可選參數as_list允許您指定對象是否作為清單傳回。

resource_fields = api.model('Resource', {
    'name': fields.String,
})

@api.route('/my-resource/<id>', endpoint='my-resource')
class MyResource(Resource):
    @api.marshal_with(resource_fields, as_list=True)
    def get(self):
        return get_objects()

    @api.marshal_with(resource_fields, code=201)
    def post(self):
        return create_object(), 201           

複制

Api.marshal_list_with()

裝飾器嚴格等價于

Api.marshal_with(fields, as_list=True)()

resource_fields = api.model('Resource', {
    'name': fields.String,
})

@api.route('/my-resource/<id>', endpoint='my-resource')
class MyResource(Resource):
    @api.marshal_list_with(resource_fields)
    def get(self):
        return get_objects()

    @api.marshal_with(resource_fields)
    def post(self):
        return create_object()           

複制

使用示例

post 請求校驗請求入參,get 請求查詢資料序列化輸出内容

from flask_restx import Resource, fields

user_input = api.model('UserModel', {
    'username': fields.String(required=True),
    'password': fields.String(required=True)
})

out_fields = api.model('UserInfo', {
    'username': fields.String,
    'email': fields.String,
    'create_time': fields.DateTime(dt_format='rfc822')
})

@api.route('/api/user', endpoint='user')
class UserView(Resource):

    @api.marshal_with(out_fields, envelope='users')
    def get(self):
        """查詢全部"""
        users = models.Users.query.all()
        return users

    @api.expect(user_input, validate=True)
    def post(self):
        """add user"""
        print(f'請求參數: {api.payload}')
        # 儲存資料庫
        return {"msg": "create success"}           

複制

get請求查詢結果

GET http://127.0.0.1:5000/api/user HTTP/1.1

{
    "users": [
        {
            "username": "test",
            "email": null,
            "create_time": "Mon, 05 Sep 2022 11:13:16 -0000"
        },
        {
            "username": "test1",
            "email": null,
            "create_time": "Mon, 05 Sep 2022 13:10:41 -0000"
        },
        {
            "username": "test12",
            "email": null,
            "create_time": "Mon, 05 Sep 2022 13:10:55 -0000"
        }
    ]
}           

複制

user_input 和 out_fields 模型可以合并為一個,有些不需要校驗,但是可以輸出的用readonly=True表示

from flask_restx import Resource, fields

user_model = api.model('UserModel', {
    'id': fields.Integer(readonly=True),
    'username': fields.String(required=True),
    'password': fields.String(required=True),
    'is_active': fields.Boolean(),
    'email': fields.String(),
    'create_time': fields.DateTime(dt_format='rfc822'),
    'update_time': fields.DateTime(dt_format='rfc822')
})

@api.route('/api/user', endpoint='user')
class UserView(Resource):

    @api.marshal_with(user_model, envelope='users')
    def get(self):
        """查詢全部"""
        users = models.Users.query.all()
        return users

    @api.expect(user_model, validate=True)
    def post(self):
        """add user"""
        print(f'請求參數: {api.payload}')
        # 儲存資料庫
        return {"msg": "create success"}           

複制

2022年第 12期《python接口web自動化+測試開發》課程,9月17号開學!

本期上課時間:2022年9月17号 - 2022年12月17号,周六周日上午9:00-11:00

報名費:報名費3000一人(周期3個月)

聯系微信/QQ:283340479