如果有幾個原因可以讓你愛上flask這個極其靈活的庫,我想藍圖絕對應該算上一個,部署藍圖以后,你會發現整個程序結構非常清晰,模塊之間相互不影響。藍圖對restful api的最明顯效果就是版本控制;而對整個項目來說,總要有后台管理系統吧,總要有web管理吧,但這些東西不能全部放到view.py。不單單是這樣,如果你是一個經驗豐富的程序員,你應該知道,一個程序最好只有一個入口點,從這個入口點進去,全是單向的,就像一棵樹一樣,入口點就在樹根,然后蔓延到樹干,樹枝。樹枝和樹枝之間最好不要太多交集,也就是我們通常所說的 低耦合。
好了,說了這么多,我們舉個簡單例子看看。假設你的一款app已經上架,現在公司要開發新的版本,新版本不但要增加新的接口,可能還要修改以前老的接口。但如果修改老的接口,那老版本的用戶又不能正常訪問了,要是用戶更新不及時,是不是這些接口就不能訪問了?如果我需要這個接口可以維持一陣子,過一段時間后,再停用,該如何實現呢?
下面這個截圖是我原先的項目結構。

如果要增加接口和修改接口,只能在view.py里面做,我這個只是舉例,在真實的項目中,一個中等app,最少有上百個接口,如何快速找到接口,定位問題,是一門很好的學問。

首先,先在app下增加一個文件夾,api_1_0作為版本1.0存放地方,是我們的一個樹枝,把之前的view.py遷移到里面去。結構如下:

我們之前總是在view.py里面運行app,這個肯定不行,不能把入口點放在某一個樹枝吧。要把整個app拉出來,在上一級來啟動項目,那么就再增加一個run.py文件,專門來運行整個項目,整個run.py代碼如下
# coding:utf-8 from flask import Flask from config import Conf import redis from qiniu import Auth, put_file, etag, urlsafe_base64_encode def create_app(): app = Flask(__name__) app.config.from_object(Conf) app.secret_key = app.config['SECRET_KEY'] app.redis = redis.Redis(host=app.config['REDIS_HOST'], port=app.config['REDIS_PORT'], db=app.config['REDIS_DB'], password=app.config['REDIS_PASSWORD']) app.q = Auth(access_key=app.config['QINIU_ACCESS_KEY'], secret_key=app.config['QINIU_SECRET_KEY']) app.bucket_name = app.config['BUCKET_NAME'] app.debug = app.config['DEBUG'] return app if __name__ == '__main__': app = create_app() app.run(debug=app.debug, host='0.0.0.0', port=5001)
這邊就是整個項目的總起點,是整個樹的根(其實這邊還可以再把run.py放在外面,這邊接下來會繼續講)。
恩,看起來好了,但我們還是沒有看到藍圖。好了,這邊就說到了,之前view.py里面全部用的app.route,之前沒問題,因為它本身就是根,可是你現在已經不是根了,那就要變換一下,在api_1_0/__init__.py里,添加如下代碼:
# coding:utf-8 from flask import Blueprint api = Blueprint('api', __name__) from . import view
這邊解釋一下,首先定義一個藍圖,這個不用講了,你用之前,肯定要定義一下吧,藍圖名字就叫api。下面一行代碼有點意思,from . import view, 這個代碼,我發現用藍圖的人,很多都忘寫了。如果你是一個python老手,你應該知道,導入python文件,其實就是運行導入的那個文件。在這邊,你導入一次,就是就是告訴上級,這邊還有個view.py文件,里面覆蓋了一些接口。整個根就知道,哦,如果有接口訪問這,我就指向你這邊。好好理解上面的粗體字。
好了,樹枝這邊已經定義了,那樹根那邊也應該重新寫一下。一個request到樹根那,樹根根據url來指向樹枝,樹根怎么指向呢?代碼如下:
def create_app(): app = Flask(__name__) app.config.from_object(Conf) app.secret_key = app.config['SECRET_KEY'] app.redis = redis.Redis(host=app.config['REDIS_HOST'], port=app.config['REDIS_PORT'], db=app.config['REDIS_DB'], password=app.config['REDIS_PASSWORD']) app.q = Auth(access_key=app.config['QINIU_ACCESS_KEY'], secret_key=app.config['QINIU_SECRET_KEY']) app.bucket_name = app.config['BUCKET_NAME'] app.debug = app.config['DEBUG'] from .app_1_0 import api as api_1_0_blueprint app.register_blueprint(api_1_0_blueprint, url_prefix='/api/v1000') return app
看到增加的2行代碼沒有,這個就是樹根指向樹枝的代碼。
好了,我們現在整個流程應該清楚了,請求從樹根進來,樹根根據藍圖指向某一個樹枝,這個樹枝具體處理請求,返回。
那自然,最后一段樹枝具體處理,也要改變咯。
原先view.py代碼改變一下,如下:
# coding:utf-8 from flask import Flask, request, jsonify, g, render_template, redirect, url_for, session, current_app from app.model import User, db_session, MaintainRecord import hashlib import time import uuid from app.util import message_validate import random from functools import wraps from . import api def login_check(f): @wraps(f) def decorator(*args, **kwargs): token = request.headers.get('token') if not token: return jsonify({'code': 0, 'message': '需要驗證'}) phone_number = current_app.redis.get('token:%s' % token) if not phone_number or token != current_app.redis.hget('user:%s' % phone_number, 'token'): return jsonify({'code': 2, 'message': '驗證信息錯誤'}) return f(*args, **kwargs) return decorator @api.before_request def before_request(): token = request.headers.get('token') phone_number = current_app.redis.get('token:%s' % token) if phone_number: g.current_user = User.query.filter_by(phone_number=phone_number).first() g.token = token return @api.route('/login', methods=['POST']) def login(): phone_number = request.get_json().get('phone_number') password = request.get_json().get('password') user = User.query.filter_by(phone_number=phone_number).first() if not user: return jsonify({'code': 0, 'message': '沒有此用戶'}) if user.password != password: return jsonify({'code': 0, 'message': '密碼錯誤'}) m = hashlib.md5() m.update(phone_number) m.update(password) m.update(str(int(time.time()))) token = m.hexdigest() pipeline = current_app.redis.pipeline() pipeline.hmset('user:%s' % user.phone_number, {'token': token, 'nickname': user.nickname, 'app_online': 1}) pipeline.set('token:%s' % token, user.phone_number) pipeline.expire('token:%s' % token, 3600*24*30) pipeline.execute() return jsonify({'code': 1, 'message': '成功登錄', 'nickname': user.nickname, 'token': token}) @api.route('/user') @login_check def user(): user = g.current_user nickname = current_app.redis.hget('user:%s' % user.phone_number, 'nickname') return jsonify({'code': 1, 'nickname': nickname, 'phone_number': user.phone_number}) @api.route('/logout') @login_check def logout(): user = g.current_user pipeline = current_app.redis.pipeline() pipeline.delete('token:%s' % g.token) pipeline.hmset('user:%s' % user.phone_number, {'app_online': 0}) pipeline.execute() return jsonify({'code': 1, 'message': '成功注銷'}) @api.route('/get-qiniu-token') def get_qiniu_token(): key = uuid.uuid4() token = current_app.q.upload_token(current_app.bucket_name, key, 3600) return jsonify({'code': 1, 'key': key, 'token': token}) @api.route('/set-head-picture', methods=['POST']) @login_check def set_head_picture(): head_picture = request.get_json().get('head_picture') user = g.current_user user.head_picture = head_picture try: db_session.commit() except Exception as e: print e db_session.rollback() return jsonify({'code': 0, 'message': '未能成功上傳'}) current_app.redis.hset('user:%s' % user.phone_number, 'head_picture', head_picture) return jsonify({'code': 1, 'message': '成功上傳'}) @api.route('/register-step-1', methods=['POST']) def register_step_1(): """ 接受phone_number,發送短信 """ phone_number = request.get_json().get('phone_number') user = User.query.filter_by(phone_number=phone_number).first() if user: return jsonify({'code': 0, 'message': '該用戶已經存在,注冊失敗'}) validate_number = str(random.randint(100000, 1000000)) result, err_message = message_validate(phone_number, validate_number) if not result: return jsonify({'code': 0, 'message': err_message}) pipeline = current_app.redis.pipeline() pipeline.set('validate:%s' % phone_number, validate_number) pipeline.expire('validate:%s' % phone_number, 60) pipeline.execute() return jsonify({'code': 1, 'message': '發送成功'}) @api.route('/register-step-2', methods=['POST']) def register_step_2(): """ 驗證短信接口 """ phone_number = request.get_json().get('phone_number') validate_number = request.get_json().get('validate_number') validate_number_in_redis = current_app.redis.get('validate:%s' % phone_number) if validate_number != validate_number_in_redis: return jsonify({'code': 0, 'message': '驗證沒有通過'}) pipe_line = current_app.redis.pipeline() pipe_line.set('is_validate:%s' % phone_number, '1') pipe_line.expire('is_validate:%s' % phone_number, 120) pipe_line.execute() return jsonify({'code': 1, 'message': '短信驗證通過'}) @api.route('/register-step-3', methods=['POST']) def register_step_3(): """ 密碼提交 """ phone_number = request.get_json().get('phone_number') password = request.get_json().get('password') password_confirm = request.get_json().get('password_confirm') if len(password) < 7 or len(password) > 30: # 這邊可以自己拓展條件 return jsonify({'code': 0, 'message': '密碼長度不符合要求'}) if password != password_confirm: return jsonify({'code': 0, 'message': '密碼和密碼確認不一致'}) is_validate = current_app.redis.get('is_validate:%s' % phone_number) if is_validate != '1': return jsonify({'code': 0, 'message': '驗證碼沒有通過'}) pipeline = current_app.redis.pipeline() pipeline.hset('register:%s' % phone_number, 'password', password) pipeline.expire('register:%s' % phone_number, 120) pipeline.execute() return jsonify({'code': 1, 'message': '提交密碼成功'}) @api.route('/register-step-4', methods=['POST']) def register_step_4(): """ 基本資料提交 """ phone_number = request.get_json().get('phone_number') nickname = request.get_json().get('nickname') is_validate = current_app.redis.get('is_validate:%s' % phone_number) if is_validate != '1': return jsonify({'code': 0, 'message': '驗證碼沒有通過'}) password = current_app.redis.hget('register:%s' % phone_number, 'password') new_user = User(phone_number=phone_number, password=password, nickname=nickname) db_session.add(new_user) try: db_session.commit() except Exception as e: print e db_session.rollback() return jsonify({'code': 0, 'message': '注冊失敗'}) finally: current_app.redis.delete('is_validate:%s' % phone_number) current_app.redis.delete('register:%s' % phone_number) return jsonify({'code': 1, 'message': '注冊成功'}) @api.teardown_request def handle_teardown_request(exception): db_session.remove()
這邊有好幾個需要注意的點。
第一,運行的代碼取消掉了,因為統一從run.py來運行,作為入口點。
第二,原先的app.route也全部改成api.route, api也從本地的__init__.py中導入。因為你現在代表樹枝,不能代表整棵樹了。
第三,app.redis,可以用current_app.redis來代替,其實就是我在run.py中定義的一些變量,在整顆樹中使用。
好了,我們運行一下python run.py試試看吧。這邊需要注意的是,url也不一樣了哦,client.py中,要用不同的url了。client.py運行代碼如下:
if __name__ == '__main__': api = APITest('http://127.0.0.1:5001/api/v1000') u = api.login('13565208554', '123456') print u u1 = api.user() print u1 api.logout()
python run.py返回的url,看看詳細情況。

整個運行情況問題,那么回到最上面提到的問題。老板看我們第一個版本做的非常穩定,要開發第二個版本,先不管新增的功能,老板覺得你上一個版本的login接口寫的不好,直接明文把用戶名和密碼傳輸過去了,安全問題得不到保障。現在要求是login接口重寫,其他不變,並且上個版本的接口暫時還不能停止。
如果按照我們以前的想法,再增加一個不同的接口唄。嗯!方法不錯,要是要修改20個接口呢?重寫嗎?怎么分得清哪個對應哪個版本?就算這次分清了,再更新5個版本,你還知道原來寫的什么嗎?
好了,上面一堆廢話就是為了襯托項目結構的重要性的。版本1.1,我們重新寫一個樹枝,這個樹枝跟1.0的樹枝沒有任何交集。有了1.0,1.1就非常好寫了。
把整個api_1_0直接復制到api_1_1中。沒錯,你沒看錯,就是整個復制過去。
然后把api_1_1中的__init__.py修改一下,代碼如下:
# coding:utf-8 from flask import Blueprint api = Blueprint('api1_1', __name__) from . import view
其實就是把藍圖名字修改了一下。整個項目結構如下圖:

在run.py中,增加如下代碼。
def create_app(): app = Flask(__name__) app.config.from_object(Conf) app.secret_key = app.config['SECRET_KEY'] app.redis = redis.Redis(host=app.config['REDIS_HOST'], port=app.config['REDIS_PORT'], db=app.config['REDIS_DB'], password=app.config['REDIS_PASSWORD']) app.q = Auth(access_key=app.config['QINIU_ACCESS_KEY'], secret_key=app.config['QINIU_SECRET_KEY']) app.bucket_name = app.config['BUCKET_NAME'] app.debug = app.config['DEBUG'] from app_1_0 import api as api_1_0_blueprint app.register_blueprint(api_1_0_blueprint, url_prefix='/api/v1000') from api_1_1 import api as api_1_1_blueprint app.register_blueprint(api_1_1_blueprint, url_prefix='/api/v1100') return app
好了,整個復制過程完工,看看實際情況吧。
我們先不修改登錄接口,就看一下效果。client.py運行代碼
if __name__ == '__main__': api = APITest('http://127.0.0.1:5001/api/v1000') u = api.login('13565208554', '123456') print u u1 = api.user() print u1 api.logout() api = APITest('http://127.0.0.1:5001/api/v1100') u = api.login('13565208554', '123456') print u u1 = api.user() print u1 api.logout()
看看ide調試結果吧。

是不是2個版本同時支持了?結構是不是非常非常清晰?
結構調整到這,那就聽老板的意思,開始寫新的接口吧。老板說,上次登錄接口不安全,要修改一下。一般來說,驗證時,都是把用戶名,和 密碼+隨機值+時間戳 的加密方式傳過去,我們也蕭規曹隨吧,新版本的login接口代碼如下:
@api.route('/login', methods=['POST']) def login(): phone_number = request.get_json().get('phone_number') encryption_str = request.get_json().get('encryption_str') random_str = request.get_json().get('random_str') time_stamp = request.get_json().get('time_stamp') user = User.query.filter_by(phone_number=phone_number).first() if not user: return jsonify({'code': 0, 'message': '沒有此用戶'}) password_in_sql = user.password s = hashlib.sha256() s.update(password_in_sql) s.update(random_str) s.update(time_stamp) server_encryption_str = s.hexdigest() if server_encryption_str != encryption_str: return jsonify({'code': 0, 'message': '密碼錯誤'}) m = hashlib.md5() m.update(phone_number) m.update(user.password) m.update(str(int(time.time()))) token = m.hexdigest() pipeline = current_app.redis.pipeline() pipeline.hmset('user:%s' % user.phone_number, {'token': token, 'nickname': user.nickname, 'app_online': 1}) pipeline.set('token:%s' % token, user.phone_number) pipeline.expire('token:%s' % token, 3600*24*30) pipeline.execute() return jsonify({'code': 1, 'message': '成功登錄', 'nickname': user.nickname, 'token': token})
代碼稍微解釋一下,encryption_str就是加密串,是由密碼+隨機值+時間戳用sha256加密的。傳到服務器,服務器也這樣加密一下,然后看看2者是不是一致。傳輸過程不涉及密碼傳輸。就這么簡單。
客戶端就不能用之前的代碼了,需要重寫一下。代碼如下:
# coding:utf-8 import requests import json from qiniu import put_file import time import random import hashlib class API_1_0(object): base_url = 'http://127.0.0.1:5001/api/v1000' def __init__(self): self.headers = {} self.token = None self.qiniu_token = None self.qiniu_key = None self.qiniu_base_url = 'http://7xk6rc.com1.z0.glb.clouddn.com/' def login(self, phone_number, password, path='/login'): payload = {'phone_number': phone_number, 'password': password} self.headers = {'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) self.token = response_data.get('token') return response_data def user(self, path='/user'): self.headers = {'token': self.token} response = requests.get(url=self.base_url + path, headers=self.headers) response_data = json.loads(response.content) return response_data def logout(self, path='/logout'): self.headers = {'token': self.token} response = requests.get(url=self.base_url + path, headers=self.headers) response_data = json.loads(response.content) return response_data def get_qiniu_token(self, path='/get-qiniu-token'): response = requests.get(url=self.base_url + path) response_data = json.loads(response.content) self.qiniu_token = response_data.get('token') self.qiniu_key = response_data.get('key') if self.qiniu_token and self.qiniu_key: print '成功獲取qiniu_token和qiniu_key,分別為%s和%s' % (self.qiniu_token.encode('utf-8'), self.qiniu_key.encode('utf-8')) localfile = '/home/yudahai/PycharmProjects/blog01/app/my-test.png' ret, info = put_file(self.qiniu_token, self.qiniu_key, localfile) print info.status_code if info.status_code == 200: print '上傳成功' self.head_picture = self.qiniu_base_url + self.qiniu_key print '其url為:' + self.head_picture.encode('utf-8') else: print '上傳失敗' return response_data def set_head_picture(self, path='/set-head-picture'): payload = {'head_picture': self.head_picture} self.headers = {'token': self.token, 'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) print response_data.get('message') return response_data def register_step_1(self, phone_number, path='/register-step-1'): payload = {'phone_number': phone_number} self.headers = {'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) print response_data.get('code') return response_data def register_step_2(self, phone_number, validate_number, path='/register-step-2'): payload = {'phone_number': phone_number, 'validate_number': validate_number} self.headers = {'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) print response_data.get('code') return response_data def register_step_3(self, phone_number, password, password_confirm, path='/register-step-3'): payload = {'phone_number': phone_number, 'password': password, 'password_confirm': password_confirm} self.headers = {'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) print response_data.get('code') return response_data def register_step_4(self, phone_number, nickname, path='/register-step-4'): payload = {'phone_number': phone_number, 'nickname': nickname} self.headers = {'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) print response_data.get('code') return response_data class API_1_1(object): base_url = 'http://127.0.0.1:5001/api/v1100' def __init__(self): self.headers = {} self.token = None self.qiniu_token = None self.qiniu_key = None self.qiniu_base_url = 'http://7xk6rc.com1.z0.glb.clouddn.com/' def login(self, phone_number, password, path='/login'): random_str = str(random.randint(10000, 100000)) time_stamp = str(int(time.time())) s = hashlib.sha256() s.update(password) s.update(random_str) s.update(time_stamp) encryption_str = s.hexdigest() payload = {'phone_number': phone_number, 'encryption_str': encryption_str, 'random_str': random_str, 'time_stamp': time_stamp} self.headers = {'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) self.token = response_data.get('token') return response_data def user(self, path='/user'): self.headers = {'token': self.token} response = requests.get(url=self.base_url + path, headers=self.headers) response_data = json.loads(response.content) return response_data def logout(self, path='/logout'): self.headers = {'token': self.token} response = requests.get(url=self.base_url + path, headers=self.headers) response_data = json.loads(response.content) return response_data def get_qiniu_token(self, path='/get-qiniu-token'): response = requests.get(url=self.base_url + path) response_data = json.loads(response.content) self.qiniu_token = response_data.get('token') self.qiniu_key = response_data.get('key') if self.qiniu_token and self.qiniu_key: print '成功獲取qiniu_token和qiniu_key,分別為%s和%s' % (self.qiniu_token.encode('utf-8'), self.qiniu_key.encode('utf-8')) localfile = '/home/yudahai/PycharmProjects/blog01/app/my-test.png' ret, info = put_file(self.qiniu_token, self.qiniu_key, localfile) print info.status_code if info.status_code == 200: print '上傳成功' self.head_picture = self.qiniu_base_url + self.qiniu_key print '其url為:' + self.head_picture.encode('utf-8') else: print '上傳失敗' return response_data def set_head_picture(self, path='/set-head-picture'): payload = {'head_picture': self.head_picture} self.headers = {'token': self.token, 'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) print response_data.get('message') return response_data def register_step_1(self, phone_number, path='/register-step-1'): payload = {'phone_number': phone_number} self.headers = {'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) print response_data.get('code') return response_data def register_step_2(self, phone_number, validate_number, path='/register-step-2'): payload = {'phone_number': phone_number, 'validate_number': validate_number} self.headers = {'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) print response_data.get('code') return response_data def register_step_3(self, phone_number, password, password_confirm, path='/register-step-3'): payload = {'phone_number': phone_number, 'password': password, 'password_confirm': password_confirm} self.headers = {'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) print response_data.get('code') return response_data def register_step_4(self, phone_number, nickname, path='/register-step-4'): payload = {'phone_number': phone_number, 'nickname': nickname} self.headers = {'content-type': 'application/json'} response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers) response_data = json.loads(response.content) print response_data.get('code') return response_data if __name__ == '__main__': api = API_1_0() u = api.login('13565208554', '123456') print u u1 = api.user() print u1 api.logout() api = API_1_1() u = api.login('13565208554', '123456') print u u1 = api.user() print u1 api.logout()
就是復制一遍,然后把base_url定義成不同,僅此而已。其實真是的ios, android也這樣。當需要更新版本,他們復制之前的項目,改變一下base_url,然后修改需要修改的接口。這樣2個版本同時支持的服務器端就寫好了,是不是超級直觀?直接運行一下吧。
{u'message': u'\u6210\u529f\u767b\u5f55', u'code': 1, u'nickname': u'\u6d4b\u8bd5\u7528\u62371', u'token': u'e5a74170d81feb106a066f18f9f2e966'}
{u'phone_number': u'13565208554', u'code': 1, u'nickname': u'\u6d4b\u8bd5\u7528\u62371'}
{u'message': u'\u6210\u529f\u767b\u5f55', u'code': 1, u'nickname': u'\u6d4b\u8bd5\u7528\u62371', u'token': u'e5a74170d81feb106a066f18f9f2e966'}
{u'phone_number': u'13565208554', u'code': 1, u'nickname': u'\u6d4b\u8bd5\u7528\u62371'}
返回的code都是1,兩種方式登錄都支持。老板要求都達到了,老板高興,發獎金,我們也高興(好吧,不做夢了)。
好了,講到這,大家應該對整個項目接口有非常清晰的了解了,以后再加接口或者更新版本,對你來說,都是不是問題了吧。
