Python使用Flask實現RESTful API,使用Postman工具、requests庫測試接口


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/<int:user_id>', 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/<int:user_id>', 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/<int:user_id>', 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/<int:id>', 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。

 

 點擊Send按鈕,Postman返回碼:Status:201 CREATED,返回內容:

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

(2)測試/api/v1.0/users/<int:user_id>的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/<int:user_id>的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/<int:user_id>的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)

 


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM