通用技术 测试开发——Flask 入门教程系列:05-规范代码结构

弗多 · February 28, 2018 · 1889 hits

传送门:系列教程前言及整理贴 测试开发——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
@login_manager.user_loader
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 项目已建成,你可以运行测试下~

共收到 0 条回复 时间 点赞
需要 Sign In 后方可回复, 如果你还没有账号请点击这里 Sign Up