编写 TODO 应用【part001】

本书前面两个部分分别对 Flask 的基本知识、用法以及介绍了多种扩展以及扩展的通用使用方式，使用扩展过程中的一些细节进行了讲解。虽然过程中有一个 REST API 小例子描述，但是，毕竟是作为各个扩展使用讲解而编排在一块，所以缺乏系统性，全局性。

从本章开始，我将使用 Flask 围绕一个 TODO 应用提供 REST API 进行讲解，让大家有个对 Flask 应用有一个直观的认识。

TODO 应用讲解

我们需要编写的 TODO 应用主要功能有：

  • 可以查询所有待办事项
  • 可以查看指定待办事项的详情
  • 可以增加一项待办事项
  • 可以删除一项待办事项
  • 可以修改一项待办事项，包括待办内容，添加标记
  • 完成待办事项后可以标记为完成

这些就是我们应用的简略需求，然后再讲一下我们的项目结构，根据前面章节《更好得维护代码》中讲解的，我们将项目结构设计成如下：

  1. .
  2. ├── README.md
  3. ├── application
  4. │   ├── __init__.py
  5. │   ├── controllers
  6. │   │   ├── __init__.py
  7. │   │   ├── auth.py
  8. │   │   ├── todo.py
  9. │   │   └── user.py
  10. │   ├── extensions.py
  11. │   └── models
  12. │       ├── __init__.py
  13. │       ├── todo.py
  14. │       └── user.py
  15. ├── commands.py
  16. ├── config
  17. │   ├── __init__.py
  18. │   ├── default.py
  19. │   ├── development.py
  20. │   ├── development_sample.py
  21. │   ├── production.py
  22. │   ├── production_sample.py
  23. │   └── testing.py
  24. ├── deploy
  25. │   ├── flask_env.sh
  26. │   ├── gunicorn.conf
  27. │   ├── nginx.conf
  28. │   └── supervisor.conf
  29. ├── manage.py
  30. ├── pylintrc
  31. ├── requirements.txt
  32. ├── tests
  33. │   └── __init__.py
  34. └── wsgi.py

设计 Models

Model 的话主要设计两个主要的模型，分别是 User 和 Item。User 表示用户的信息，除了表示TODO 所属人之外，还有登录的用处，而 Item 则是待办事项了，具体设计需要参考需求而定，关于 Model 的具体设计过程不是本章讨论的重点，所以直接给出 Models：

application/models/init.py

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. from user import *
  4. from todo import *
  7. def all():
  8.     result = []
  9.     models = [user, todo]
  11.     for m in models:
  12.         result += m.__all__
  14.     return result
  16. __all__ = all()

application/models/user.py

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. from application.extensions import db
  5. __all__ = ['Role', 'User']
  8. class Permission:
  9.     READ = 0x01
  10.     CREATE = 0x02
  11.     UPDATE = 0x04
  12.     DELETE = 0x08
  13.     DEFAULT = READ
  16. class Role(db.Document):
  17.     name = db.StringField()
  18.     permission = db.IntField()
  21. class User(db.Document):
  22.     name = db.StringField()
  23.     password = db.StringField()
  24.     email = db.StringField()
  25.     role = db.ReferenceField('Role')
  27.     def to_json(self):
  28.         return {"name": self.name,
  29.                 "email": self.email,
  30.                 "role": self.role.name}
  32.     def is_authenticated(self):
  33.         return True
  35.     def is_active(self):
  36.         return True
  38.     def is_anonymous(self):
  39.         return False
  41.     def get_id(self):
  42.         return str(self.id)

application/models/todo.py

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. from application.extensions import db
  5. __all__ = ['Item']
  8. class Item(db.Document):
  9.     content = db.StringField(required=True)
  10.     created_date = db.DateTimeField()
  11.     completed = db.BooleanField(default=False)
  12.     completed_date = db.DateTimeField()
  13.     created_by = db.ReferenceField('User', required=True)
  14.     notes = db.ListField(db.StringField())
  15.     priority = db.IntField()
  17.     def __repr__(self):
  18.         return "<Item: {} Content: {}>".format(str(self.id),
  19.                                                self.content)
  21.     def to_json(self):
  22.         return {
  23.             'id': str(self.id),
  24.             'content': self.content,
  25.             'completed': self.completed,
  26.             'completed_at': self.completed_date.strftime("%Y-%m-%d %H:%M:%S") if self.completed else "",
  27.             'created_by': self.created_by.name,
  28.             'notes': self.notes,
  29.             'priority': self.priority
  30.         }

设计 views

根据我们在前面章节所学习的知识，我们这个应用的 views 就不是直接使用 app.route 来绑定 URL 了，而是使用 Blueprint 来设计，具体设计如下：

application/controller/init.py

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. import auth
  4. import user
  5. import todo
  7. all_bp = [
  8.     auth.auth_bp,
  9.     user.user_bp,
  10.     todo.todo_bp
  11. ]

application/controller/auth.py

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. import json
  5. from flask import Blueprint, request, jsonify
  6. from flask.ext.login import login_user, logout_user
  8. import application.models as Models
  11. auth_bp = Blueprint('auth', __name__, url_prefix='/auth')
  14. @auth_bp.route('/login', methods=['POST'])
  15. def login():
  16.     info = json.loads(request.data)
  17.     username = info.get('username', 'guest')
  18.     password = info.get('password', '')
  20.     user = Models.User.objects(name=username,
  21.                                password=password).first()
  22.     if user:
  23.         login_user(user)
  24.         return jsonify(user.to_json())
  25.     else:
  26.         return jsonify({"status": 401,
  27.                         "reason": "Username or Password Error"})
  30. @auth_bp.route('/logout', methods=['POST'])
  31. def logout():
  32.     logout_user()
  33.     return jsonify(**{'result': 200,
  34.                       'data': {'message': 'logout success'}})

application/controller/user.py

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. from flask.ext.login import current_user
  4. from flask import Blueprint, jsonify
  7. user_bp = Blueprint('users', __name__, url_prefix='')
  10. @user_bp.route('/user_info', methods=['POST'])
  11. def user_info():
  12.     if current_user.is_authenticated:
  13.         resp = {"result": 200,
  14.                 "data": current_user.to_json()}
  15.     else:
  16.         resp = {"result": 401,
  17.                 "data": {"message": "user no login"}}
  18.     return jsonify(**resp)

application/controller/todo.py

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. import json
  4. from datetime import datetime
  6. from flask import Blueprint, request, jsonify
  7. from flask.ext.login import current_user, login_required
  9. import application.models as Models
  12. todo_bp = Blueprint('todos', __name__, url_prefix='/todo')
  15. @todo_bp.route('/item', methods=['POST'])
  16. @login_required
  17. def create_todo_item():
  18.     data = json.loads(request.data)
  19.     content = data.get('content')
  20.     note = data.get('note', None)
  21.     priority = data.get('priority', 0)
  23.     if not content:
  24.         return jsonify({
  25.             'data': {},
  26.             'msg': 'no content',
  27.             'code': 1001,
  28.             'extra': {}})
  30.     item = Models.Item(content=content, created_date=datetime.now(),
  31.                        completed=False, created_by=current_user.id,
  32.                        notes=[note] if note else [],
  33.                        priority=priority)
  34.     item.save()
  35.     return jsonify({
  36.         'data': item.to_json(),
  37.         'msg': 'create item success',
  38.         'code': 1000,
  39.         'extra': {}
  40.     })
  43. @todo_bp.route('/item', methods=['DELETE'])
  44. @login_required
  45. def delete_todo_item():
  46.     data = json.loads(request.data)
  47.     id = data.get('id')
  49.     if not id:
  50.         return jsonify({
  51.             'data': {},
  52.             'msg': 'no id',
  53.             'code': 2001,
  54.             'extra': {}})
  56.     item = Models.Item.objects(id=id).first()
  57.     item.delete()
  58.     return jsonify({
  59.         'data': item.to_json(),
  60.         'msg': 'delete item success',
  61.         'code': 2000,
  62.         'extra': {}
  63.     })
  66. @todo_bp.route('/item', methods=['PUT'])
  67. @login_required
  68. def update_todo_item():
  69.     data = json.loads(request.data)
  70.     id = data.get('id')
  71.     type = data.get('type')
  73.     if type == "update_content":
  74.         content = data.get('content')
  75.         Models.Item.objects(id=id).first().update(content=content)
  76.     elif type == "insert_notes":
  77.         note = data.get('note')
  78.         Models.Item.objects(id=id).first().update(push__notes=note)
  79.     elif type == "done":
  80.         Models.Item.objects(id=id).first().update(completed=True,
  81.                                                   completed_date=datetime.now())
  82.     return jsonify({
  83.         'data': {'oper': type,
  84.                  'id': id},
  85.         'msg': 'oper done',
  86.         'code': 3000,
  87.         'extra': {}
  88.     })
  91. @todo_bp.route('/item', methods=['GET'])
  92. @login_required
  93. def get_todo_item():
  94.     query_string = request.args.get('q')
  95.     data = json.loads(query_string)
  96.     id = data.get('id')
  98.     item = Models.Item.objects(id=id).first()
  99.     return jsonify({
  100.         'data': item.to_json(),
  101.         'msg': 'query item success',
  102.         'code': 4000,
  103.         'extra': {}
  104.     })
  107. @todo_bp.route('/items', methods=['GET'])
  108. @login_required
  109. def get_todo_items():
  110.     data = json.loads(request.args.get('q'))
  111.     page = data.get('page', 1)
  112.     page_size = data.get('page_size', 10)
  114.     begin = (page - 1) * page_size
  115.     end = begin + page_size
  116.     items = Models.Item.objects()[begin: end]
  117.     rsts = []
  118.     for item in items:
  119.         rsts.append(item.to_json())
  121.     return jsonify({
  122.         'data': rsts,
  123.         'msg': 'query items success',
  124.         'code': 5000,
  125.         'extra': {}
  126.     })

初始化扩展

扩展我们是统一放到 application/extensions.py 里面进行构建对象的，所以文件有：

application/extensions.py

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. from flask.ext.admin import Admin
  4. from flask.ext.login import LoginManager
  5. from flask.ext.mongoengine import MongoEngine
  8. db = MongoEngine()
  9. login_manager = LoginManager()
  10. admin = Admin()

初始化应用

  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. import sys
  4. import logging
  6. from flask import Flask
  7. from flask_admin.contrib.mongoengine import ModelView
  9. from config import load_config
  10. from application.extensions import db, login_manager, admin
  11. from application.models import User, Role
  12. from application.controllers import all_bp
  14. # convert python's encoding to utf8
  15. try:
  16.     reload(sys)
  17.     sys.setdefaultencoding('utf8')
  18. except (AttributeError, NameError):
  19.     pass
  22. def create_app(mode):
  23.     """Create Flask app."""
  24.     config = load_config(mode)
  26.     app = Flask(__name__)
  27.     app.config.from_object(config)
  29.     if not hasattr(app, 'production'):
  30.         app.production = not app.debug and not app.testing
  32.     if app.debug or app.testing:
  33.         # Log errors to stderr in production mode
  34.         app.logger.addHandler(logging.StreamHandler())
  35.         app.logger.setLevel(logging.ERROR)
  37.     # Register components
  38.     register_extensions(app)
  39.     register_blueprint(app)
  41.     return app
  44. def register_extensions(app):
  45.     """Register models."""
  46.     db.init_app(app)
  47.     login_manager.init_app(app)
  49.     # flask-admin configs
  50.     admin.init_app(app)
  51.     admin.add_view(ModelView(User))
  52.     admin.add_view(ModelView(Role))
  54.     login_manager.login_view = 'auth.login'
  56.     @login_manager.user_loader
  57.     def load_user(user_id):
  58.         return User.objects(id=user_id).first()
  61. def register_blueprint(app):
  62.     for bp in all_bp:
  63.         app.register_blueprint(bp)