上一章,咱們講到,怎麼用藍圖建造一個好的項目,今天咱們繼續深刻。上一章中,咱們全部的接口都寫在view.py中,若是幾十個,還稍微好管理一點,假如上百個,上千個,怎麼找?全部接口堆在一塊兒就顯得雜亂無章。flask沒有推薦你們在這方面的功能,一般都是由本身來實現。咱們一般的作法,都是按照功能劃分文件,把不一樣功能的應用接口,劃分到不一樣文件,若是某個功能的接口不少,再細分一下。固然你也能夠按照其餘劃分方式劃分,只要記住一點,項目怎麼管理方便,就怎麼劃分,千萬不要被框架框死。html
廢話說過了,來點實際的。上一個版本,咱們只有註冊,登陸,註銷等基本功能,總不能就修改一個加密接口吧,這一個版本老闆要求把微博功能加進去,固然,這邊的微博不是新浪的那個,就是能發一些東西,裏面有標題,不超過140字的文字內容,不超過9張圖片的圖片集等。繼續在上一個版本的model.py和view.py中添加。python
model.py以下:redis
class SmallBlog(Base): __tablename__ = 'small_blog' id = Column('id', Integer, primary_key=True) post_user_id = Column('post_user_id', Integer, ForeignKey(User.id)) post_time = Column('post_time', DateTime, default=datetime.datetime.now) title = Column('title', String(30), index=True) text_content = Column('text_content', String(140)) picture_content = Column('picture_content', String(900)) post_user = relationship('User', backref=backref('small_blogs')) @hybrid_property def pictures(self): if not self.picture_content: return [] return self.picture_content.split(',') @pictures.setter def pictures(self, urls): self.picture_content = ','.join(urls) def to_dict(self): return { 'id': self.id, 'post_user_picture': self.post_user.head_picture, 'post_user_name': self.post_user.nickname, 'post_time': self.post_time.strftime('%Y-%m-%d %H:%M:%S'), 'title': self.title, 'text_content': self.text_content, 'pictures': self.pictures }
就多了一個SmallBlog的model,而後更新數據庫,如何更新數據庫,請看個人 flask開發restful api系列(3)--利用alembic進行數據庫更改 這篇文章。還有,這邊的圖片存儲方式,我是用每張圖片的url,中間用","分隔開,而後存儲,客戶端請求的時候,再分隔開,返回url的數組。這樣就用最簡單的方式實現了0到9張圖片的動態添加。若是你有更好的方案,請及時回覆我,你們共同窗習。sql
還有,我在這裏面定義了3個函數,關於pictures的顯示和設置,這個稍微熟悉一點python的都知道,只不過在普通的類中,直接@property。而在sqlalchemy裏面,用@hybrid_property便可。還有就是專門爲返回json格式數據作準備的函數to_dict(),獲取每一個對象,而後執行這個函數就能夠返回字典方式。這個你們能夠用別的庫,也能夠用本身寫的to_dict函數,我在不少地方強調,寫代碼必定要靈活,千萬不能被框架限定死了。若是哪天以爲很差,也能夠用flask別的擴展庫實現串行化對象。數據庫
model.py方面添加好,下面就在view.py中添加以下函數:json
view.pyflask
1 @api.route('/get-multi-qiniu-token') 2 @login_check 3 def get_multi_qiniu_token(): 4 count = request.args.get('count') 5 6 if not 0 < int(count) < 10: 7 return jsonify({'code': 0, 'message': '一次只能獲取1到9個'}) 8 9 key_token_s = [] 10 for x in range(int(count)): 11 key = uuid.uuid1() 12 token = current_app.q.upload_token(current_app.bucket_name, key, 3600) 13 key_token_s.append((key, token)) 14 return jsonify({'code': 1, 'key_token_s': key_token_s}) 15 16 17 @api.route('/post-blog', methods=['POST']) 18 @login_check 19 def post_blog(): 20 user = g.current_user 21 22 title = request.get_json().get('title') 23 text_content = request.get_json().get('text_content') 24 pictures = request.get_json().get('pictures') 25 26 newblog = SmallBlog(title=title, text_content=text_content, post_user=user) 27 28 newblog.pictures = pictures 29 db_session.add(newblog) 30 try: 31 db_session.commit() 32 except Exception as e: 33 print e 34 db_session.rollback() 35 return jsonify({'code': 0, 'message': '上傳不成功'}) 36 return jsonify({'code': 1, 'message': '上傳成功'}) 37 38 39 @api.route('/get-blogs') 40 @login_check 41 def get_blogs(): 42 last_id = request.args.get('last_id') 43 if not int(last_id): 44 blogs = db_session.query(SmallBlog).order_by(desc(SmallBlog.id)).limit(10) 45 else: 46 blogs = db_session.query(SmallBlog).filter(SmallBlog.id < int(last_id)).order_by(desc(SmallBlog.id)).limit(10) 47 return jsonify({'code': 1, 'blogs': [blog.to_dict() for blog in blogs]})
多了3個函數,第一個函數get_multi_qiniu_token,看名字也能夠看出來,這個就是獲取七牛token的方法,只不過獲取多個,這邊限制在1到9個之間,本身上傳一個數目,而後返回key和token。api
第二個函數 post_blog就是一個提交新blog的過程,惟一要注意的就是pictures,在客戶端直接上傳數組,而後賦值就能夠了。數組
第三個函數 get_blogs有點意思,其實就是根據last_id獲得上面10條數據,若是第一次沒有last_id,就填0就能夠了。最主要是給客戶端使用,你總不能一會兒把全部數據都給客戶端,一般都是必定的條目,而後下拉,根據最上面的id,來獲取其上10條數據,這樣能夠無限下拉刷新,只要一個接口就行了。就跟微博同樣。若是你有其餘好辦法,歡迎一塊兒探討。服務器
服務器端寫好了,咱們就寫客戶端吧,客戶端代碼也很簡單,針對這3個函數,寫3個接口而已。
client.py
def get_multi_qiniu_token(self, count, path='/get-multi-qiniu-token'): self.headers = {'token': self.token} payload = {'count': count} response = requests.get(url=self.base_url + path, params=payload, headers=self.headers) response_data = json.loads(response.content) key_token_s = response_data.get('key_token_s') return key_token_s def post_blog(self, title, text_content, picture_files, path='/post-blog'): self.headers = {'token': self.token} count = len(picture_files) key_token_s = self.get_multi_qiniu_token(count=count) pictures = [] for x in range(count): put_file(key_token_s[x][1], key_token_s[x][0], picture_files[x]) pictures.append(self.qiniu_base_url + key_token_s[x][0]) payload = {'title': title, 'text_content': text_content, 'pictures': pictures} self.headers = {'content-type': 'application/json', 'token': self.token} 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 get_blogs(self, last_id, path='/get-blogs'): self.headers = {'token': self.token} payload = {'last_id': last_id} response = requests.get(url=self.base_url + path, params=payload, headers=self.headers) response_data = json.loads(response.content) return response_data
測試一下吧,
if __name__ == '__main__': api = API_1_1() u = api.login('13565208554', '123456') # key_token_s = api.get_multi_qiniu_token(4) api.post_blog(title=u'dsfsdfrerg434', text_content=u'是打發撒dsfsdgsdfsgs法上得分未違法', picture_files=['./img/4.png', './img/2.png', './img/3.png']) blogs = api.get_blogs(0) print blogs api.logout()
隨便上傳幾個,打印一下,就能夠看到效果了,好了,添加完畢。
下面咱們就介紹一下結構,想象一下,咱們目前添加了最最基本的功能,就要添加3個接口。其實一個功能慢慢作下去,幾十個接口都是少的,若是都放在view.py裏面,你們想象一個文件有多少接口,不要說看起來很是難看,就是之後查找bug,也不是通常的複雜。在傳統python中,一般都是根據功能劃分文件,固然也有其餘的,就像咱們上面所說的,只要找到適合你本身的,就去改變,不要被框架限定死。
咱們目前在view.py裏面,其實有不少功能了,有裝飾器模塊,驗證模塊,註冊模塊,七牛模塊,博客模塊等,其實如今接口少,就少分一點文件,這個只要靈活便可,沒有必要限定本身,咱們就劃分auth.py, main.py, decorators.py,blogs.py這4個模塊,其中auth.py裏面包含驗證和註冊;main.py裏面包含一些公用,包括運行接口先後以及七牛token和key的獲取;decorators.py就是裝飾器;blogs.py就是博客接口。大體先這麼分,之後遇到其餘接口,再動態添加便可。
轉移一下以前的代碼,把view.py的代碼分別複製到各個py文件下,而後刪除view.py文件,如今的文件結構以下:
是否是清晰多了?這裏再把各個py文件的代碼複製一下,最最重要的就是__init__.py裏面的代碼,必定要覆蓋到全部接口。
# coding:utf-8 from flask import Blueprint api = Blueprint('api1_1', __name__) from . import auth, blogs, decorators, main
看清楚from . import auth, blogs, decorators, main這行代碼,就是程序啓動的時候,覆蓋全部接口。
下面分別是auth.py, blogs.py, decorators.py, main.py文件,這裏就不用一一介紹了吧。
auth.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, SmallBlog, desc import hashlib import time from app.util import message_validate import random from .decorators import login_check from . import api @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}) @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('/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': '註冊成功'})
blogs.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, SmallBlog, desc from .decorators import login_check from . import api @api.route('/post-blog', methods=['POST']) @login_check def post_blog(): user = g.current_user title = request.get_json().get('title') text_content = request.get_json().get('text_content') pictures = request.get_json().get('pictures') newblog = SmallBlog(title=title, text_content=text_content, post_user=user) newblog.pictures = pictures db_session.add(newblog) try: db_session.commit() except Exception as e: print e db_session.rollback() return jsonify({'code': 0, 'message': '上傳不成功'}) return jsonify({'code': 1, 'message': '上傳成功'}) @api.route('/get-blogs') @login_check def get_blogs(): last_id = request.args.get('last_id') if not int(last_id): blogs = db_session.query(SmallBlog).order_by(desc(SmallBlog.id)).limit(10) else: blogs = db_session.query(SmallBlog).filter(SmallBlog.id < int(last_id)).order_by(desc(SmallBlog.id)).limit(10) return jsonify({'code': 1, 'blogs': [blog.to_dict() for blog in blogs]})
decorator.py
# coding:utf-8 from flask import Flask, request, jsonify, g, render_template, redirect, url_for, session, current_app from functools import wraps 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
main.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, SmallBlog, desc import uuid from . import api from .decorators import login_check @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.teardown_request def handle_teardown_request(exception): db_session.remove() @api.route('/get-multi-qiniu-token') @login_check def get_multi_qiniu_token(): count = request.args.get('count') if not 0 < int(count) < 10: return jsonify({'code': 0, 'message': '一次只能獲取1到9個'}) key_token_s = [] for x in range(int(count)): key = uuid.uuid1() token = current_app.q.upload_token(current_app.bucket_name, key, 3600) key_token_s.append((key, token)) return jsonify({'code': 1, 'key_token_s': key_token_s}) @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})
運行一下,看看有沒有問題。整個文件結構進一步優化,我想哪一個接口在哪一個文件裏,一目瞭然。其實如今能夠回頭看看,假設哪天個人model裏面有上百張表,我該如何用model.py文件呢?這個能夠留給你們去實踐一下,必定要記住,這雖然不是必要的,但好的項目結構能讓人一會兒明白你整個項目,即便之後你離職,下一個同事也能快速上手。最後,再囉嗦一句,必定要靈活,不要被框架限定死。