Python使用Flask实现RESTful API,使用Postman工具

系统 1794 0

RESTful是一种API设计规范。
在RESTful架构中,主要使用POST,DELETE,PUT和GET四种HTTP请求方式分别对指定的URL资源进行增删改查操作。

RESTful之前的做法:

            /users/query/1 GET 根据用户id查询用户数据
/users/save POST 新增用户
/users/update POST 修改用户信息
/users/delete GET/POST 删除用户信息
          

RESTful做法:

            /users/1 GET 根据用户id查询用户数据
/users POST 新增用户
/users PUT 修改用户信息
/users DELETE 删除用户信息
          

客户端的每一次请求,服务器都会给出回应,回应包括 HTTP 状态码和数据两部分。
部分状态码:
GET: 200 OK
POST: 201 Created
PUT: 200 OK
DELETE: 204 No Content

 

使用Flask实现一个RESTful API服务的例子

            
              from
            
             flask 
            
              import
            
            
               Flask,jsonify,abort,make_response,request

            
            
              from
            
             flask_httpauth 
            
              import
            
            
               HTTPBasicAuth

app 
            
            = Flask(
            
              __name__
            
            
              )
auth 
            
            =
            
               HTTPBasicAuth()

users 
            
            =
            
               [
    {
        
            
            
              '
            
            
              id
            
            
              '
            
            : 1
            
              ,
        
            
            
              '
            
            
              username
            
            
              '
            
            : 
            
              '
            
            
              小明
            
            
              '
            
            
              ,
        
            
            
              '
            
            
              sex
            
            
              '
            
            : 1
            
              
    },
    {
        
            
            
              '
            
            
              id
            
            
              '
            
            : 2
            
              ,
        
            
            
              '
            
            
              username
            
            
              '
            
            : 
            
              '
            
            
              小红
            
            
              '
            
            
              ,
        
            
            
              '
            
            
              sex
            
            
              '
            
            
              : 0
    }
]


            
            
              #
            
            
              访问前需要提供用户名和密码
            
            
              @auth.get_password

            
            
              def
            
            
               get_password(username):
    
            
            
              if
            
             username == 
            
              '
            
            
              admin
            
            
              '
            
            
              :
        
            
            
              return
            
            
              '
            
            
              123456
            
            
              '
            
            
              return
            
            
               None


            
            
              #
            
            
              友好的错误提示:没有权限
            
            
              @auth.error_handler

            
            
              def
            
            
               unauthorized():
    
            
            
              return
            
             make_response(jsonify({
            
              '
            
            
              error
            
            
              '
            
            : 
            
              '
            
            
              未授权
            
            
              '
            
            }), 403
            
              )


            
            
              #
            
            
              友好的错误提示:找不到资源页面
            
            
@app.errorhandler(404
            
              )

            
            
              def
            
            
               not_found(error):
    
            
            
              return
            
             make_response(jsonify({
            
              '
            
            
              error
            
            
              '
            
            : 
            
              '
            
            
              找不到资源
            
            
              '
            
            }), 404
            
              )


            
            
              #
            
            
              返回所有用户的记录
            
            
@app.route(
            
              '
            
            
              /api/v1.0/users
            
            
              '
            
            , methods=[
            
              '
            
            
              GET
            
            
              '
            
            
              ])
@auth.login_required 
            
            
              #
            
            
              需要认证
            
            
              def
            
            
               get_users():
    
            
            
              return
            
             jsonify({
            
              '
            
            
              users
            
            
              '
            
            
              : users})


            
            
              #
            
            
              返回一个用户的记录记录
            
            
@app.route(
            
              '
            
            
              /api/v1.0/users/
              
            
            
              '
            
            , methods=[
            
              '
            
            
              GET
            
            
              '
            
            
              ])
@auth.login_required

            
            
              def
            
            
               get_user(user_id):
    user 
            
            = list(u 
            
              for
            
             u 
            
              in
            
             users 
            
              if
            
             u[
            
              '
            
            
              id
            
            
              '
            
            ] ==
            
               user_id)
    
            
            
              if
            
             len(user) ==
            
               0:
        abort(
            
            404
            
              )
    
            
            
              return
            
             jsonify({
            
              '
            
            
              user
            
            
              '
            
            
              : user[0]})


            
            
              #
            
            
              插入一条用户记录
            
            
@app.route(
            
              '
            
            
              /api/v1.0/users
            
            
              '
            
            , methods=[
            
              '
            
            
              POST
            
            
              '
            
            
              ])
@auth.login_required

            
            
              def
            
            
               create_user():
    
            
            
              if
            
            
              not
            
             request.json 
            
              or
            
            
              not
            
            
              '
            
            
              username
            
            
              '
            
            
              in
            
             request.json 
            
              or
            
            
              not
            
            
              '
            
            
              sex
            
            
              '
            
            
              in
            
            
               request.json:
        abort(
            
            400
            
              )
    user 
            
            =
            
               {
        
            
            
              '
            
            
              id
            
            
              '
            
            : users[-1][
            
              '
            
            
              id
            
            
              '
            
            ] + 1
            
              ,
        
            
            
              '
            
            
              username
            
            
              '
            
            : request.json[
            
              '
            
            
              username
            
            
              '
            
            
              ],
        
            
            
              '
            
            
              sex
            
            
              '
            
            : request.json[
            
              '
            
            
              sex
            
            
              '
            
            
              ]
    }
    users.append(user)
    
            
            
              return
            
             jsonify({
            
              '
            
            
              user
            
            
              '
            
            : user}), 201


            
              #
            
            
              更新一个用户的记录
            
            
@app.route(
            
              '
            
            
              /api/v1.0/users/
              
            
            
              '
            
            , methods=[
            
              '
            
            
              PUT
            
            
              '
            
            
              ])
@auth.login_required

            
            
              def
            
            
               update_user(user_id):
    user 
            
            = list(u 
            
              for
            
             u 
            
              in
            
             users 
            
              if
            
             u[
            
              '
            
            
              id
            
            
              '
            
            ] ==
            
               user_id)
    
            
            
              if
            
             len(user) ==
            
               0:
        abort(
            
            404
            
              )
    
            
            
              if
            
            
              not
            
            
               request.json:
        abort(
            
            400
            
              )   
    user[0][
            
            
              '
            
            
              username
            
            
              '
            
            ] = request.json.get(
            
              '
            
            
              username
            
            
              '
            
            , user[0][
            
              '
            
            
              username
            
            
              '
            
            
              ])
    user[0][
            
            
              '
            
            
              sex
            
            
              '
            
            ] = request.json.get(
            
              '
            
            
              sex
            
            
              '
            
            , user[0][
            
              '
            
            
              sex
            
            
              '
            
            
              ])
    
            
            
              return
            
             jsonify({
            
              '
            
            
              user
            
            
              '
            
            
              : user[0]})


            
            
              #
            
            
              删除一个用户的记录
            
            
@app.route(
            
              '
            
            
              /api/v1.0/users/
              
            
            
              '
            
            , methods=[
            
              '
            
            
              DELETE
            
            
              '
            
            
              ])
@auth.login_required

            
            
              def
            
            
               delete_user(user_id):
    user 
            
            = list(u 
            
              for
            
             u 
            
              in
            
             users 
            
              if
            
             u[
            
              '
            
            
              id
            
            
              '
            
            ] ==
            
               user_id)
    
            
            
              if
            
             len(user) ==
            
               0:
        abort(
            
            404
            
              )
    users.remove(user[0])
    
            
            
              return
            
             jsonify({
            
              '
            
            
              result
            
            
              '
            
            : True}),204



            
              if
            
            
              __name__
            
             == 
            
              '
            
            
              __main__
            
            
              '
            
            
              :
    app.run(debug
            
            =True)
          

 

上面例子,也可以使用Flask-RESTful实现

            
              from
            
             flask 
            
              import
            
            
               Flask,jsonify,abort,make_response,request

            
            
              from
            
             flask_httpauth 
            
              import
            
            
               HTTPBasicAuth

            
            
              from
            
             flask_restful 
            
              import
            
            
               Api,Resource,reqparse

app 
            
            = Flask(
            
              __name__
            
            
              )
api 
            
            =
            
               Api(app)
auth 
            
            =
            
               HTTPBasicAuth()

users 
            
            =
            
               [
    {
        
            
            
              '
            
            
              id
            
            
              '
            
            : 1
            
              ,
        
            
            
              '
            
            
              username
            
            
              '
            
            : 
            
              '
            
            
              小明
            
            
              '
            
            
              ,
        
            
            
              '
            
            
              sex
            
            
              '
            
            : 1
            
              
    },
    {
        
            
            
              '
            
            
              id
            
            
              '
            
            : 2
            
              ,
        
            
            
              '
            
            
              username
            
            
              '
            
            : 
            
              '
            
            
              小红
            
            
              '
            
            
              ,
        
            
            
              '
            
            
              sex
            
            
              '
            
            
              : 0
    }
]


            
            
              #
            
            
              验证字段的合法性
            
            
parser =
            
               reqparse.RequestParser()
parser.add_argument(
            
            
              '
            
            
              username
            
            
              '
            
            , type = str, required = True, help = 
            
              '
            
            
              此字段格式有问题
            
            
              '
            
            , location = 
            
              '
            
            
              json
            
            
              '
            
            
              )
parser.add_argument(
            
            
              '
            
            
              sex
            
            
              '
            
            , type = int, default = 
            
              "
            
            
              1
            
            
              "
            
            , location = 
            
              '
            
            
              json
            
            
              '
            
            
              )  


            
            
              #
            
            
              访问前需要提供用户名和密码
            
            
              @auth.get_password

            
            
              def
            
            
               get_password(username):
    
            
            
              if
            
             username == 
            
              '
            
            
              admin
            
            
              '
            
            
              :
        
            
            
              return
            
            
              '
            
            
              123456
            
            
              '
            
            
              return
            
            
               None


            
            
              class
            
            
               UserListAPI(Resource):
    decorators 
            
            =
            
               [auth.login_required]   

    
            
            
              def
            
            
               get(self):
        
            
            
              return
            
             {
            
              '
            
            
              users
            
            
              '
            
            
              : users}

    
            
            
              def
            
            
               post(self):
        args 
            
            =
            
               parser.parse_args()
        user 
            
            =
            
               {        
            
            
            
              '
            
            
              id
            
            
              '
            
            : users[-1][
            
              '
            
            
              id
            
            
              '
            
            ] + 1
            
              ,
            
            
            
              '
            
            
              username
            
            
              '
            
            : args[
            
              '
            
            
              username
            
            
              '
            
            
              ],
            
            
            
              '
            
            
              sex
            
            
              '
            
            : args[
            
              '
            
            
              sex
            
            
              '
            
            
              ]
        }
        users.append(user)
        
            
            
              return
            
             {
            
              '
            
            
              user
            
            
              '
            
            : user}, 201


            
              class
            
            
               UserAPI(Resource):
    decorators 
            
            =
            
               [auth.login_required]    

    
            
            
              def
            
            
               get(self, id):        
        user 
            
            = list(u 
            
              for
            
             u 
            
              in
            
             users 
            
              if
            
             u[
            
              '
            
            
              id
            
            
              '
            
            ] ==
            
               id)
        
            
            
              if
            
             len(user) ==
            
               0:
            abort(
            
            404
            
              )
        
            
            
              return
            
             {
            
              '
            
            
              user
            
            
              '
            
            
              : user[0]}

    
            
            
              def
            
            
               put(self, id):
        user 
            
            = list(u 
            
              for
            
             u 
            
              in
            
             users 
            
              if
            
             u[
            
              '
            
            
              id
            
            
              '
            
            ] ==
            
               id)
        
            
            
              if
            
             len(user) ==
            
               0:
            abort(
            
            404
            
              )
        user 
            
            =
            
               user[0]
        args 
            
            =
            
               parser.parse_args()
        
            
            
              for
            
             k, v 
            
              in
            
            
               args.items():
            
            
            
              if
            
             v !=
            
               None:
                user[k] 
            
            =
            
               v
        
            
            
              return
            
              { 
            
              '
            
            
              user
            
            
              '
            
            
              : user }

    
            
            
              def
            
            
               delete(self, id):
        user 
            
            = list(u 
            
              for
            
             u 
            
              in
            
             users 
            
              if
            
             u[
            
              '
            
            
              id
            
            
              '
            
            ] ==
            
               id)
        
            
            
              if
            
             len(user) ==
            
               0:
            abort(
            
            404
            
              )
        users.remove(user[0])
        
            
            
              return
            
             {
            
              '
            
            
              result
            
            
              '
            
            
              : True}

api.add_resource(UserListAPI, 
            
            
              '
            
            
              /api/v1.0/users
            
            
              '
            
            , endpoint = 
            
              '
            
            
              users
            
            
              '
            
            
              )
api.add_resource(UserAPI, 
            
            
              '
            
            
              /api/v1.0/users/
              
            
            
              '
            
            , endpoint = 
            
              '
            
            
              user
            
            
              '
            
            
              )


            
            
              if
            
            
              __name__
            
             == 
            
              '
            
            
              __main__
            
            
              '
            
            
              :
    app.run(debug
            
            =True)
          

 

使用Postman测试接口

创建一个新的Request,Request name和Create Collection都随便输入如RESTfulAPI,保存后,自动打开名称为RESTfulAPI的标签页。

(1)测试/api/v1.0/users的GET请求

RESTfulAPI标签页默认选择 GET,输入:localhost:5000/api/v1.0/users,这里直接点击Send按钮,Postman返回码:Status:403 FORBIDDEN,返回内容

            {
"error": "未授权"
}
          

在Authorization标签里面TYPE选择Basic Auth,右边Username和Password分别输入admin和123456。

Python使用Flask实现RESTful API,使用Postman工具、requests库测试接口_第1张图片

 

 点击Send按钮,Postman返回码:Status:201 CREATED,返回内容:

            {
  "users": [
    {
      "id": 1,
      "sex": 1,
      "username": "小明"
    },
    {
      "id": 2,
      "sex": 0,
      "username": "小红"
    }
  ]
}
          

(2)测试/api/v1.0/users/ 的GET请求

修改Postman的请求方法和地址为:GET localhost:5000/api/v1.0/users/2

(3)测试/api/v1.0/users的POST请求

修改Postman的请求方法和地址为:POST localhost:5000/api/v1.0/users
在Body标签里面选择raw、修改最后的默认Text为JSON(application/json),下面输入框填写:

            {
	"id": 3,
	"sex": 1,
	"username": "小强"
}
          

点击Send按钮,Postman返回码:Status:201 CREATED
把POST方法直接改为GET方法,可见返回3条记录。

(4)测试/api/v1.0/users/ 的PUT请求

修改Postman的请求方法和地址为:PUT localhost:5000/api/v1.0/users/3
在Body标签里面选择raw、修改最后的默认Text为JSON(application/json),下面输入框填写:

            {
	"id": 3,
	"sex": 1,
	"username": "小刚"
}
          

点击Send按钮,Postman返回码:Status:200 OK

(5)测试/api/v1.0/users/ 的DELETE请求

修改Postman的请求方法和地址为:DELETE localhost:5000/api/v1.0/users/3
点击Send按钮,Postman返回码:Status:204 NOT CONTENT

 

使用Python的第三方库requests测试接口

            
              import
            
            
               requests,json

url 
            
            = 
            
              '
            
            
              http://localhost:5000/api/v1.0
            
            
              '
            
            
              
auth 
            
            = (
            
              '
            
            
              admin
            
            
              '
            
            ,
            
              '
            
            
              123456
            
            
              '
            
            
              )


            
            
              #
            
            
              查询
            
            
r=requests.get(url + 
            
              '
            
            
              /users
            
            
              '
            
            , auth=
            
              auth)

            
            
              print
            
            
              (r.status_code, r.text)

            
            
              #
            
            
              在控制台输出时中文会用unicode显示,可用下面方法显示中文
            
            
              
#
            
            
              print(json.dumps(json.loads(r.text),ensure_ascii=False))
            
            
              
r
            
            =requests.get(url + 
            
              '
            
            
              /users/1
            
            
              '
            
            , auth=
            
              auth)

            
            
              print
            
            
              (r.status_code, r.text)


            
            
              #
            
            
              更新
            
            
data =
            
               {
    
            
            
              "
            
            
              id
            
            
              "
            
            : 3
            
              ,
    
            
            
              "
            
            
              sex
            
            
              "
            
            : 1
            
              ,
    
            
            
              "
            
            
              username
            
            
              "
            
            : 
            
              "
            
            
              小强
            
            
              "
            
            
              
}
r
            
            =requests.post(url + 
            
              '
            
            
              /users
            
            
              '
            
            , auth=auth, json=
            
              data)

            
            
              print
            
            
              (r.status_code, r.text)


            
            
              #
            
            
              修改
            
            
data =
            
               {
    
            
            
              "
            
            
              id
            
            
              "
            
            : 3
            
              ,
    
            
            
              "
            
            
              sex
            
            
              "
            
            : 1
            
              ,
    
            
            
              "
            
            
              username
            
            
              "
            
            : 
            
              "
            
            
              小刚
            
            
              "
            
            
              
}
r
            
            =requests.put(url + 
            
              '
            
            
              /users/3
            
            
              '
            
            , auth=auth, json=
            
              data)

            
            
              print
            
            
              (r.status_code, r.text)


            
            
              #
            
            
              删除
            
            
r=requests.delete(url + 
            
              '
            
            
              /users/3
            
            
              '
            
            , auth=
            
              auth)

            
            
              print
            
            (r.status_code, r.text)
          

 


更多文章、技术交流、商务合作、联系博主

微信扫码或搜索:z360901061

微信扫一扫加我为好友

QQ号联系: 360901061

您的支持是博主写作最大的动力,如果您喜欢我的文章,感觉我的文章对您有帮助,请用微信扫描下面二维码支持博主2元、5元、10元、20元等您想捐的金额吧,狠狠点击下面给点支持吧,站长非常感激您!手机微信长按不能支付解决办法:请将微信支付二维码保存到相册,切换到微信,然后点击微信右上角扫一扫功能,选择支付二维码完成支付。

【本文对您有帮助就好】

您的支持是博主写作最大的动力,如果您喜欢我的文章,感觉我的文章对您有帮助,请用微信扫描上面二维码支持博主2元、5元、10元、自定义金额等您想捐的金额吧,站长会非常 感谢您的哦!!!

发表我的评论
最新评论 总共0条评论