通過Flask實現Restful服務
阿新 • • 發佈：2019-01-05
1，前言
Python的強大，已經涉及到軟體開發領域的方方面面。然而，Python入門容易，精確很難，需要深入研究。
在Web方面同樣如此，常用的Python Web框架，例如Django、Flask、Tornado等等，共計有100多種，各有優劣。本文以Flask為例，介紹Flask的Restful實現方式，主要實現對資料表的增刪查改操作。
2，需求
在開發程式碼之前，需要提前明確URL請求、HTTP方法與對應的動作。
2.1，Mysql資料表
建立一個簡單的資料表，表名是student，儲存學生的name、age、birth_time資訊。
2.2，學生列表的操作
URL - get 獲取學生列表
- post 建立新學生資訊
2.3，單個學生的操作
URL：http://127.0.0.1:5002/student/<student_id>- get 獲取單個學生資訊
- put 更新單個學生資訊
- delete 刪除單個學生資訊
2.4，附屬功能
- 要求每次請求之前，記錄使用者的IP、請求引數等資訊，便於統計分析
- 要求每次資料庫操作，判斷資料庫連線是否正常，避免長時間未進行DB操作而造成DB連線失連
3，Flask專案
3.1，建立Flask專案
- Windows可以使用PyCharm安裝，或者手動安裝
- Ubuntu安裝flask，執行 sudo apt-get install python-flask
- 建立hello.py，測試flask是否OK
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run() - 啟動flask，執行 python hello.py
- 測試flask，執行 curl http://127.0.0.1:5000/，發現返回“Hello World!”，證明安裝成功
- 如果返回缺少某些python包，莫急，逐個安裝即可
3.2，目錄規劃
按照MVC的思想，對控制層和檢視層進行分離。控制層主要是對資料庫的處理，可以按照不同的資料表進行分檔案處理；檢視層，按照不同的功能模組，進行份檔案處理。這樣，便於專案維護和模組擴充套件。
不過，一定還有更好的目錄規劃方式，因專案或團隊不同而不同，適合自己的即是最好的。
下面是初步劃分的檔案目錄，可以參考。
D:\PYTEST\FLASK_TEST
│ ctrl_student.py # 檢視層，響應URL請求
│ db_student.py # 控制層，資料庫處理
│ flask_test.py # Flask主檔案，控制URL路由
│ test.py # 測試檔案
├─static
└─templates4，原始碼
4.1，資料庫表
- student表結構
CREATE TABLE `student` (
`id` int(10) unsigned NOT NULL AUTO_INCREMENT,
`name` varchar(255) DEFAULT NULL,
`age` int(11) DEFAULT NULL,
`birth_time` datetime DEFAULT NULL,
`create_time` datetime DEFAULT NULL,
`is_del` int(10) unsigned zerofill DEFAULT '0000000000',
PRIMARY KEY (`id`)
) ENGINE=MyISAM AUTO_INCREMENT=19 DEFAULT CHARSET=utf8;4.2，Python原始碼
- flask_test.py
# coding:utf8
from flask import Flask, render_template, request
from flask_restful import Api
from ctrl_student import StudentList, Student
app = Flask(__name__)
api = Api(app)
api.add_resource(StudentList, '/student/')
api.add_resource(Student, '/student/<student_id>')
@app.before_request
def before_request():
ip = request.remote_addr
url = request.url
form = request.form # 請求的資料，可執行searchword = request.form.get('key', '') ?????????測試（帶引數的post請求）過程中form為空，不清楚原因
args = request.args # ?key=value，可執行searchword = request.args.get('key', '')
values = request.values # form和args的元組
headers = request.headers
method = request.method
path = request.path
base_url = request.base_url
url_root = request.url_root
print "ip", ip
print "url", url
# print "form", form
# print "args", args
# print "values", values
# print "headers", headers
# print "method", method
# print "path", path
# print "base_url", base_url
# print "url_root", url_root
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=5002)- ctrl_student.py
# coding:utf8
from flask_restful import reqparse, abort, Api, Resource
from db_student import DbStudent
# 404操作
def abort_if_todo_doesnt_exist(student_id):
if DbStudent.get_student_by_id(student_id) == {}:
abort(404, message="Todo {} doesn't exist".format(student_id))
# 入參解析器
parser = reqparse.RequestParser()
# parser.add_argument('id')
parser.add_argument('name')
parser.add_argument('age')
parser.add_argument('birth_time')
# 學生資訊操作
class Student(Resource):
def __init__(self):
pass
# 獲取student資訊
def get(self, student_id):
abort_if_todo_doesnt_exist(student_id)
res = DbStudent.get_student_by_id(student_id)
return res, 200
# 更新student資訊
def put(self, student_id):
print "put come in"
abort_if_todo_doesnt_exist(student_id)
args = parser.parse_args()
print args
DbStudent.update_student_by_id(student_id, args)
return {"status":"success"}, 200
# 刪除student資訊
def delete(self, student_id):
print "put come in"
abort_if_todo_doesnt_exist(student_id)
DbStudent.delete_student_by_id(student_id)
return {"status":"success"}, 200
# 相容post操作，獲取student資訊
def post(self, student_id):
print "post come in"
return self.get(student_id)
# 學生列表操作
class StudentList(Resource):
# 獲取學生資訊列表
def get(self):
res = DbStudent.get_student_list()
return res, 200
# 增加單個學生資訊
def post(self):
args = parser.parse_args()
DbStudent.insert_student(args)
return {"status":"success"}, 200- db_student.py
# coding:utf8
import MySQLdb
host = "192.168.1.64"
port = 3306
user = "******"
passwd = "******"
db = "test"
conn = None
cur = None
# 執行資料庫操作之前判斷資料庫連線是否OK，如果異常則建立一次連線
def db_opt(func):
def db_ping():
global conn, cur
try:
cur.execute("select 1")
cur.fetchone()
except:
conn = MySQLdb.connect(host=host, port=port, user=user, passwd=passwd, db=db, charset="utf8")
cur = conn.cursor()
print "build a new connection"
def wrapper(*args, **kwargs):
global conn, cur
try:
db_ping()
res = func(*args, **kwargs)
print "db_opt", func, args, res
return res
except:
traceback.print_exc('wrapper ')
return {'status':False, 'data':'db operation maybe have some errors.'}
return wrapper
class DbStudent:
# 根據student_id獲取學生資訊
@staticmethod
@db_opt
def get_student_by_id(student_id):
sql = 'select id, name, age, birth_time from student where id = %s and is_del = 0' % student_id
cur.execute(sql)
res = cur.fetchall()
if len(res) == 0:
return {}
else:
(id, name, age, birth_time) = res[0]
dict_student ={}
dict_student['id'] = id
dict_student['name'] = name
dict_student['age'] = age
dict_student['birth_time'] = str(birth_time)
return dict_student
# 根據name獲取學生資訊
@staticmethod
@db_opt
def get_student_by_name(name):
sql = 'select id, name, age, birth_time from student where name = "%s" and is_del = 0' % name
cur.execute(sql)
res = cur.fetchall()
if len(res) == 0:
return {}
else:
(id, name, age, birth_time) = res[0]
dict_student = {}
dict_student['id'] = id
dict_student['name'] = name
dict_student['age'] = age
dict_student['birth_time'] = str(birth_time)
return dict_student
# 根據student_id更新學生資訊
@staticmethod
@db_opt
def update_student_by_id(student_id, student_info):
sql = 'update student set name = "%s", age = %s, birth_time = "%s" where id = %s' % \
(student_info['name'], student_info['age'], student_info['birth_time'], student_id)
cur.execute(sql)
conn.commit
# 增加一條學生資訊
@staticmethod
@db_opt
def insert_student(student_info):
sql = 'insert into student(name, age, birth_time, create_time) values("%s", %s, "%s", now())' % \
(student_info['name'], student_info['age'], student_info['birth_time'])
cur.execute(sql)
conn.commit
# 根據student_id刪除學生資訊（更新is_del欄位）
@staticmethod
@db_opt
def delete_student_by_id(student_id):
sql = "update student set is_del = 1 where id = %s" % student_id
cur.execute(sql)
conn.commit
# 獲取學生資訊列表
@staticmethod
@db_opt
def get_student_list():
sql = 'select id, name, age, birth_time from student where is_del = 0'
cur.execute(sql)
res = cur.fetchall()
if len(res) == 0:
return {}
else:
student_list = []
for (id, name, age, birth_time) in res:
dict_student = {}
dict_student['id'] = id
dict_student['name'] = name
dict_student['age'] = age
dict_student['birth_time'] = str(birth_time)
student_list.append( dict_student )
return {"student_list": student_list}
if __name__ == '__main__':
print DbStudent.get_student_by_id(1)
print DbStudent.get_student_by_name('zhangsp')
print DbStudent.get_student_list()5，測試
curl -X GET http://192.168.1.64:5002/student/
- 增加1個學生資訊
curl -H "Content-type: application/json" -X POST -d '{"age": 3, "birth_time": "2017-08-02 18:15:42", "name": "c"}' http://192.168.1.64:5002/student/
- 更新單個學生資訊
curl -H "Content-type: application/json" -X PUT -d '{"age": 333, "birth_time": "2017-08-02 18:15:42", "name": "c"}' http://192.168.1.64:5002/student/19
- 刪除單個學生資訊
curl -H "Content-type: application/json" -X DELETE http://192.168.1.64:5002/student/19