传送门：系列教程前言及整理贴 测试开发——Flask 入门教程系列 (整理及前言)

项目目录结构规范

第三章我们学习编写了 TODO Task 相关 API，第四章我们加入了用户概念并编写了注册登录等用户相关的 API。

随着功能的不断叠加，我们的代码日渐庞大，难道要在一个 py 文件中继续写写写么？显然这是不合理的。

本章我们就之前的代码做下拆分重整，规范项目的目录结构。

如果之前有项目基础的，大多都会有自己规整代码目录结构方式，公司级别的项目更会有明确的规范。

此外，一些较大型的 Web 框架会创建默认的项目目录结构（如 Django），而 Flask 没有且官方没有规定。

HelloWorld

从零开始，我们就 HelloWorld 做下拆分，新建文件夹 05-1_Demo，其下创建如下文件：

└── 05-1_Demo
    ├── app
    │   ├── __init__.py
    │   └── helloworld.py
    └── run.py

run.py 为运行文件，内容如下：

#!/usr/bin/env python3

from app import app

if __name__ == '__main__':
    app.run(debug=True)

__init__.py 文件定义了 app：

#!/usr/bin/env python3

from flask import Flask
from flask_cors import CORS
from flask_restful import Api

app = Flask(__name__)
cors_options = {"supports_credentials": True}
cors = CORS(app, **cors_options)
api = Api(app)

from app import helloworld

附加，引入 flask-cors 解决跨域请求问题

helloworld.py 即路由及逻辑代码：

#!/usr/bin/env python3

from app import app

from flask import jsonify

@app.route('/')
def index():
    return jsonify({'msg': 'Hello World!'})

完事后，我们返回目录，可运行 python run.py 并访问 http://127.0.0.1:5000/ 测试。

TODO API 目录结构

现在我们来扩展出 TODO API 项目目录结构：

└── 05-2_Todo
    ├── app
    │   ├── __init__.py
    │   ├── models
    │   │   ├── __init__.py
    │   │   ├── task.py
    │   │   └── user.py
    │   └── routes
    │       ├── __init__.py
    │       ├── auth.py
    │       ├── task.py
    │       └── user.py
    ├── dev.py
    └── run.py

简述下一级文件夹及文件：

1. app/ 项目 APP 的代码，开发的代码都放在这
2. dev.py 为 development 环境下的配置文件
3. run.py 运行 app

后续章节会具体优化配置管理，并介绍项目部署，加入部署用 wsgi。

后续章节会引入 Flask-Script 改写 run.py

其实和 HelloWorld 只差一个 dev.py 配置，我们暂时只需要这些：

# Flask app config
DEBUG = True
TESTING = False
SECRET_KEY = "secret_key"

# MongoEngine config
MONGODB_SETTINGS = {
    'db': 'todo',
    'host': 'localhost',
    'port': 27017
}

然后，我们重点看下 app 文件夹下的目录及文件，并看下如何做代码分解的：

• 将 model 代码放到 app/models 中
• 将 route 代码放到 app/routes 中
• 将初始化绑定 app 的代码放到 app/__init__.py 中

app/__init__.py

#!/usr/bin/env python3

from flask import Flask
from flask_cors import CORS
from flask_restful import Api
from flask_mongoengine import MongoEngine
from flask_login import LoginManager

app = Flask(__name__)
cors_options = {"supports_credentials": True}
cors = CORS(app, **cors_options)
api = Api(app)

app.config.from_object('dev')

db = MongoEngine()
db.init_app(app)

# login_manager & load_user
login_manager = LoginManager()
login_manager.init_app(app)

from app.models.user import User
@user2oader
def load_user(user_id):
    return User.objects(user_id=user_id).first()

# Import routes
from app.routes import auth
from app.routes import user
from app.routes import task

具体创建了 app；引入了 db；定义了通用的 login 及 load_user；并在最后引入了对应的路由模块。

models

根据之前的代码，我们创建了两个数据模型，分别是 User 及 Task。

/models/user.py

#!/usr/bin/env python3

from app import db


class User(db.Document):
    user_id = db.IntField(required=True)
    name = db.StringField(required=True, max_length=100)
    email = db.StringField(max_length=200)
    pwd = db.StringField(requied=True, min_length=6)
    createtime = db.DateTimeField(required=True)

    def to_json(self):
        return {
            "user_id": self.user_id,
            "name": self.name,
            "email": self.email
        }

    def is_authenticated(self):
        return True

    def is_active(self):
        return True

    def is_anonymous(self):
        return False

    def get_id(self):
        return str(self.user_id)

model/task.py

#!/usr/bin/env python3

from app import db


class Task(db.Document):
    user_id = db.IntField(required=True)
    task_id = db.StringField(required=True)
    title = db.StringField(required=True, max_length=50)
    description = db.StringField(required=True, max_length=1000)
    done = db.BooleanField(required=True)
    createtime = db.DateTimeField(required=True)
    completetime = db.DateTimeField()

    def to_json(self):
        return {
            "user_id": self.user_id,
            "task_id": self.task_id,
            "title": self.title,
            "description": self.description,
            "done": self.done,
            "createtime": self.createtime.strftime("%Y-%m-%d %H:%M:%S"),
            "completetime": self.completetime.strftime("%Y-%m-%d %H:%M:%S") if self.done else ""
        }

routes

同样根据之前的代码，我们拆分并改写相应的代码，这边将注册登录登出等分离出为 auth.py。

routes/auth.py

#!/usr/bin/env python3

from app import app

from flask import jsonify, request
from flask_login import login_user, logout_user

from datetime import datetime

from app.models.user import User


@app.route('/register', methods=['POST'])
def registerUser():
    if not request.json or not 'name' in request.json or not 'pwd' in request.json:
        return jsonify({'err': 'Request not Json or miss name/pwd'})
    elif User.objects(name=request.json['name']).first():
        return jsonify({'err': 'Name is already existed.'})
    else:
        user = User(
            user_id=User.objects().count() + 1,
            name=request.json['name'],
            email=request.json['email'] if 'email' in request.json else "",
            pwd=request.json['pwd'],
            createtime=datetime.now()
        )
        try:
            user.save()
            login_user(user)
        except Exception as e:
            print(e)
            return jsonify({'err': 'Register error.'})
    return jsonify({'status': 0, 'user_id': user['user_id'], 'msg': 'Register success.'})


@app.route('/login', methods=['POST'])
def login():
    if not request.json or not 'name' in request.json or not 'pwd' in request.json:
        return jsonify({'err': 'Request not Json or miss name/pwd'})
    else:
        user = User.objects(
            name=request.json['name'], pwd=request.json['pwd']).first()
    if user:
        login_user(user)
        return jsonify({'status': 0, 'user_id': user.get_id(), 'msg': 'Login success.'})
    else:
        return jsonify({'err': 'Login fail.'})


@app.route('/logout', methods=['POST'])
def logout():
    logout_user()
    return jsonify({'status': 0, 'msg': 'Logout success.'})

routes/user.py

#!/usr/bin/env python3

from app import app

from flask import jsonify, request
from flask_login import current_user, login_required

from app.models.user import User


@app.route('/user', methods=['GET'])
def getUser():
    if current_user.is_authenticated:
        return jsonify({'status': 0, 'user': current_user.to_json()})
    else:
        return jsonify({'err': 'Not login.'})


@app.route('/user/email', methods=['PUT'])
@login_required
def putUserEmail():
    if not request.json or not 'email' in request.json:
        return jsonify({'err': 'Request not Json or miss email'})
    else:
        current_user.email = request.json['email']
        try:
            current_user.save()
        except Exception:
            return jsonify({'err': 'Modify email error.'})
        return jsonify({'status': 0, 'msg': 'Email has been modified.', 'user': current_user.to_json()})


@app.route('/user/pwd', methods=['PUT'])
@login_required
def putUserPWD():
    if not request.json or not 'current_pwd' in request.json or not 'new_pwd' in request.json:
        return jsonify({'err': 'Request not Json or miss current_pwd/new_pwd'})
    else:
        current_pwd = current_user.pwd
    if not request.json['current_pwd'] == current_pwd:
        return jsonify({'err': 'current_pwd is not right.'})
    else:
        current_user.pwd = request.json['new_pwd']
        try:
            current_user.save()
        except Exception:
            return jsonify({'err': 'Modify PWD error.'})
        return jsonify({'status': 0, 'msg': 'PWD has been modified.', 'user_id': current_user.user_id})

routes/task.py

task 的代码需要做一定的改写，之前的 task 是不区分用户的，这边加入所属用户的逻辑，且 API 均需要登录态

#!/usr/bin/env python3

from app import app

from flask import jsonify, request
from flask_login import current_user, login_required

from app.models.task import Task

from datetime import datetime
import shortuuid


@app.route('/todo/task', methods=['POST'])
@login_required
def postTask():
    if not request.json or not 'task' in request.json:
        return jsonify({'err': 'Request not Json or miss task.'})
    else:
        task = Task(
            user_id=current_user.user_id,
            task_id=shortuuid.uuid(),
            title=request.json['task'],
            description=request.json['description'] if 'decription' in request.json else "",
            done=False,
            createtime=datetime.now()
        )
        task.save()
    return jsonify({'status': 0, 'task_id': task['task_id']})


@app.route('/todo/task/<task_id>', methods=['GET'])
@login_required
def getTask(task_id):
    task = Task.objects(user_id=current_user.user_id, task_id=task_id).first()
    if not task:
        return jsonify({'err': 'Not found.'})
    else:
        return jsonify({'status': 0, 'task': task.to_json()})


@app.route('/todo/tasks', methods=['GET'])
@login_required
def getTasks():
    tasks = Task.objects(user_id=current_user.user_id)
    return jsonify({'status': 0, 'tasks': [task.to_json() for task in tasks]})


@app.route('/todo/task/<task_id>', methods=['PUT'])
@login_required
def putTask(task_id):
    task = Task.objects(user_id=current_user.user_id, task_id=task_id).first()
    if not task:
        return jsonify({'err': 'Not found.'})
    else:
        if 'task' in request.json:
            task.update(title=request.json['task'])
        if 'description' in request.json:
            task.update(description=request.json['description'])
        if 'done' in request.json:
            if request.json['done'] == True:
                task.update(done=True, completetime=datetime.now())
        task = Task.objects(task_id=task_id).first()
        return jsonify({'status': 0, 'task': task.to_json()})


@app.route('/todo/task/<task_id>', methods=['DELETE'])
@login_required
def deleteTask(task_id):
    task = Task.objects(user_id=current_user.user_id, task_id=task_id).first()
    if not task:
        return jsonify({'err': 'Not found.'})
    else:
        task.delete()
        return jsonify({'status': 0, 'task_id': task['task_id']})

至此，一个完整的基础 TODO API 项目已建成，你可以运行测试下~