flask開發restful api系列(8)-再談項目結構

時間 2019-11-17

標籤 flask 開發 restful api 系列再談項目結構欄目 Python 简体版

原文原文鏈接

　　上一章，咱們講到，怎麼用藍圖建造一個好的項目，今天咱們繼續深刻。上一章中，咱們全部的接口都寫在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': '註冊成功'})

View Code

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]})

View Code

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

View Code

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})

View Code

　　運行一下，看看有沒有問題。整個文件結構進一步優化，我想哪一個接口在哪一個文件裏，一目瞭然。其實如今能夠回頭看看，假設哪天個人model裏面有上百張表，我該如何用model.py文件呢？這個能夠留給你們去實踐一下，必定要記住，這雖然不是必要的，但好的項目結構能讓人一會兒明白你整個項目，即便之後你離職，下一個同事也能快速上手。最後，再囉嗦一句，必定要靈活，不要被框架限定死。