上一章,我们讲到,怎么用蓝图建造一个好的项目,今天我们继续深入。上一章中,我们所有的接口都写在view.py中,如果几十个,还稍微好管理一点,假如上百个,上千个,怎么找?所有接口堆在一起就显得杂乱无章。flask没有推荐大家在这方面的功能,通常都是由自己来实现。我们通常的做法,都是按照功能划分文件,把不同功能的应用接口,划分到不同文件,如果某个功能的接口很多,再细分一下。当然你也可以按照其他划分方式划分,只要记住一点,项目怎么管理方便,就怎么划分,千万不要被框架框死。

  废话说过了,来点实际的。上一个版本,我们只有注册,登录,注销等基本功能,总不能就修改一个加密接口吧,这一个版本老板要求把微博功能加进去,当然,这边的微博不是新浪的那个,就是能发一些东西,里面有标题,不超过140字的文字内容,不超过9张图片的图片集等。继续在上一个版本的model.py和view.py中添加。

model.py如下:

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张图片的动态添加。如果你有更好的方案,请及时回复我,大家共同学习。

  还有,我在这里面定义了3个函数,关于pictures的显示和设置,这个稍微熟悉一点python的都知道,只不过在普通的类中,直接@property。而在sqlalchemy里面,用@hybrid_property即可。还有就是专门为返回json格式数据做准备的函数to_dict(),获取每个对象,然后执行这个函数就可以返回字典方式。这个大家可以用别的库,也可以用自己写的to_dict函数,我在很多地方强调,写代码一定要灵活,千万不能被框架限定死了。如果哪天觉得不好,也可以用flask别的扩展库实现串行化对象。

  model.py方面添加好,下面就在view.py中添加如下函数:

view.py

 @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('/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]})

  多了3个函数,第一个函数get_multi_qiniu_token,看名字也可以看出来,这个就是获取七牛token的方法,只不过获取多个,这边限制在1到9个之间,自己上传一个数目,然后返回key和token。

  第二个函数 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(')
    # 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(')
    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)

    ':
        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)

    ':
        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文件呢?这个可以留给大家去实践一下,一定要记住,这虽然不是必要的,但好的项目结构能让人一下子明白你整个项目,即使以后你离职,下一个同事也能快速上手。最后,再啰嗦一句,一定要灵活,不要被框架限定死。

  

  

flask开发restful api系列(8)-再谈项目结构的更多相关文章

  1. flask开发restful api系列(7)-蓝图与项目结构

    如果有几个原因可以让你爱上flask这个极其灵活的库,我想蓝图绝对应该算上一个,部署蓝图以后,你会发现整个程序结构非常清晰,模块之间相互不影响.蓝图对restful api的最明显效果就是版本控制:而 ...

  2. flask开发restful api系列(1)

    在此之前,向大家说明的是,我们整个框架用的是flask + sqlalchemy + redis.如果没有开发过web,还是先去学习一下,这边只是介绍如果从开发web转换到开发移动端.如果flask还 ...

  3. flask开发restful api系列(6)-配置文件

    任何一个好的程序,配置文件必不可少,而且非常重要.配置文件里存储了连接数据库,redis的用户密码,不允许有任何闪失.要有灵活性,用户可以自己配置:生产环境和开发环境要分开,最好能简单的修改一个东西, ...

  4. flask开发restful api系列(5)-短信验证码

    我们现在开发app,注册用户的时候,不再像web一样,发送到个人邮箱了,毕竟个人邮箱在移动端填写验证都很麻烦,一般都采用短信验证码的方式.今天我们就讲讲这方面的内容. 首先,先找一个平台吧.我们公司找 ...

  5. flask开发restful api系列(4)--七牛图片服务

    上一章我们讲到如何利用alembic来更新数据库,这章,我们讲如何通过七牛服务来存储图片. 像我们大多数公司一样,公司资金比较少,如果自己开发图片服务器,代价太大:如果我们用自己的网站服务器来保存图片 ...

  6. flask开发restful api系列(3)--利用alembic进行数据库更改

    上面两章,主要讲基本的配置,今天我们来做一个比较有趣的东西,为每个客户加一个头像图片.如果我们图片保存在自己的服务器,对于服务器要求有点高,每次下载的时候,都会阻塞网络接口,要是1000个人同时访问这 ...

  7. flask开发restful api系列(2)

    继续上一章所讲,上一章我们最后面说道,虽然这个是很小的程序,但还有好几个要优化的地方.先复制一下老的view.py代码. # coding:utf-8 from flask import Flask, ...

  8. flask开发restful api

    flask开发restful api 如果有几个原因可以让你爱上flask这个极其灵活的库,我想蓝图绝对应该算上一个,部署蓝图以后,你会发现整个程序结构非常清晰,模块之间相互不影响.蓝图对restfu ...

  9. 描述怎样通过flask+redis+sqlalchemy等工具,开发restful api

    flask开发restful api系列(8)-再谈项目结构 摘要: 进一步介绍flask的项目结构,使整个项目结构一目了然.阅读全文 posted @ 2016-06-06 13:54 月儿弯弯02 ...

随机推荐

  1. Android 学习第12课,应用出错信息

    应用在运行时,出现的错误信息都会在LogCat中显示 如果调出LogCat ? 菜单:窗口 -> 显示视图 -> 其他 -> LogCat

  2. 【ASP.net】Equals 和 == 的区别

    在比较Equals 和 ==的区别前.我们先来了解下相关的知识 C#数据类型 1.值类型 值类型有: 值类型包括:简单类型.结构类型.枚举类型:引用类型包括:Object 类型.类类型.接口.代表元. ...

  3. urllib下载文件

    import urllib 1.用urlib.urlretrieve f = urllib.urlretrieve('http://www.baidu.com/img/bdlogo.gif','/tm ...

  4. bootstrap3.0 模态框显示的文字超出怎么办

    版本:bootstrap3.3 模态框文字超出 解决方案: 在html中控制自动换行   其实只要在表格控制中添加一句<div style="word-break:break-all& ...

  5. Windows NTService 后台框架封装

    对于后台运行的程序,比如基于C/S架构的服务器.各种监控系统的程序.或者是程序额外的功能需要后台运行来实现,在Windows平台中,服务经常会用到,这种对于需要24×7运行的程序是必备的,至少本人经常 ...

  6. javascript 交互取值

    var publicClassName; var classIdInMemory = { lastVal: publicClassName, set:function(x){ if(x != &quo ...

  7. SVG动画实践篇-模拟音量高低效果

    git 地址:https://github.com/rainnaZR/svg-animations/tree/master/src/demo/step2/volumn 说明 这个动画的效果就是多个线条 ...

  8. 分布式web架构中对session同步的常用处理方法以及优缺点

    写在前面 最近在读一本来自淘宝技术团队大牛的书,名字叫<大型网站系统与Java中间件实践>.开篇的章节详细地介绍了一个网站架构由小变大不断演进的过程,其中从单机架构升级到集群架构的过程中着 ...

  9. Bash : test 命令

    在 Bash 脚本中我们一般会使用 test 命令来进行条件检查.test 命令的返回值为 0 或 1.0 表示 true, 1 表示 false.简单起见,我们可以直接认为 test 的结果为 tr ...

  10. scrapy爬取小说盗墓笔记

    # -*- coding: utf-8 -*- import scrapy from daomu.items import DaomuItem class DaomuspiderSpider(scra ...