Flask项目基本拆分
垃圾桶边的狗 关注
用到的知识点
Flask四大内置对象
- request
- session
- g
- app (config)
提升代码复用性
- 封装函数
- 装饰器
- 类继承
- 钩子函数
钩子函数
- 中间件一个东西
- 面向切面编程的切点
- before_request
Django :process_reqeustafter_request
Django : process_response- errorhandler
Django : processs_exception
插件使用
- 寻找适合的插件
- 安装插件
- 初始化插件
- 操作插件
工作环境
- 开发环境
- 开发工程师使用
- 测试环境
- 测试工程师使用
- 演示环境
- 给项目经理看
- 演习使用
- 线上环境（生产环境）
- 真实用户使用的环境
流程图
[图片上传失败...(image-8d9a8b-1544252761684)]
项目目录
[图片上传失败...(image-a84813-1544252761684)]
App/init.py
from flask import Flask
from App.extension import init_ext
from App.middleware import load_middleware
from App.settings import envs
from App.views import init_blue
def create_app(env):
app = Flask(__name__)
# 初始化App的配置
app.config.from_object(envs.get(env))
# 初始化第三方的插件
init_ext(app=app)
# 加载中间件
load_middleware(app=app)
# 初始化路由系统
init_blue(app=app)
return app
App/manage.py
import os
from flask_migrate import MigrateCommand
from flask_script import Manager
from App import create_app
env = os.environ.get("FLASK_PROJECT", "product")
app = create_app(env)
manager = Manager(app=app)
manager.add_command("db", MigrateCommand)
if __name__ == '__main__':
manager.run()
App/extension.py
from flask_caching import Cache
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
migrate = Migrate()
cache = Cache(config={
"CACHE_TYPE": "redis"
})
def init_ext(app):
db.init_app(app=app)
migrate.init_app(app=app, db=db)
cache.init_app(app)
App/settings.py
知识点：
对不同的环境进行分离
def get_db_uri(dbinfo):
engine = dbinfo.get("ENGINE")
driver = dbinfo.get("DRIVER")
user = dbinfo.get("USER")
password = dbinfo.get("PASSWORD")
host = dbinfo.get("HOST")
port = dbinfo.get("PORT")
name = dbinfo.get("NAME")
return "{}+{}://{}:{}@{}:{}/{}".format(engine, driver,user, password, host, port, name)
class Config:
DEBUG = False
TESTING = False
SECRET_KEY = "wertyuiodfghjkl!@#$%^&I(*&^5"
SQLALCHEMY_TRACK_MODIFICATIONS = False
class DevelopConfig(Config):
DEBUG = True
dbinfo = {
"ENGINE": "mysql",
"DRIVER": "pymysql",
"USER": "root",
"PASSWORD": "123456",
"HOST": "localhost",
"PORT": "3306",
"NAME": "FlaskProject1"
}
SQLALCHEMY_DATABASE_URI = get_db_uri(dbinfo)
class TestingConfig(Config):
TESTING = True
class StagingConfig(Config):
pass
class ProductConfig(Config):
pass
envs = {
"develop": DevelopConfig,
"testing": TestingConfig,
"staging": StagingConfig,
"product": ProductConfig,
"default": DevelopConfig,
}
App/models/init.py
from .sutdent_model import Student
from .teacher_model import Teacher
App/models/student_model.py
知识点：
1.用户密码加密
2.用户密码校验
4.数据自动保存数据库
5.用户权限校验
from werkzeug.security import generate_password_hash, check_password_hash
from App.extension import db
PERMISSION_STUDENT_LEARN = 1
PERMISSION_STUDENT_MANAGE = 2
PERMISSION_STUDENT_USE_PHONE = 4
class Student(db.Model):
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
s_name = db.Column(db.String(16), unique=True)
_s_password = db.Column(db.String(256))
s_permission = db.Column(db.Integer, default=PERMISSION_STUDENT_LEARN)
@property
def s_password(self):
raise Exception("can't access")
@s_password.setter
def s_password(self, password):
self._s_password = generate_password_hash(password)
def verify_password(self, password):
return check_password_hash(self._s_password, password)
def check_permission(self, permission):
return self.s_permission & permission == permission
def save(self):
try:
db.session.add(self)
db.session.commit()
except Exception as e:
print(e)
return False
else:
return True
App/views/init.py
from App.views.book_blue import book_blue
from App.views.first_blue import blue
from App.views.student_blue import blue_student
def init_blue(app):
app.register_blueprint(blueprint=blue)
app.register_blueprint(blueprint=blue_student)
app.register_blueprint(blueprint=book_blue)
App/views/login_register.py
>知识点：
uuid生成token唯一值
缓存：对token进行过期时间设置
abort
import uuid
from flask import Blueprint, request, abort, jsonify
from App.extension import cache
from App.models import Student
blue_student = Blueprint("blue_student", __name__, url_prefix='/students')
@blue_student.route("/", methods=["POST", ])
def students():
action = request.args.get("action")
username = request.form.get("username")
password = request.form.get("password")
if action == "register":
student = Student()
student.s_name = username
student.s_password = password
if not student.save():
abort(401)
return "注册成功"
elif action == "login":
# Student.query.filter(Student.s_name.__eq__(username))
student_list = Student.query.filter(Student.s_name == (username)).all()
if not student_list:
abort(404)
student = student_list[0]
if not student.verify_password(password):
abort(401)
token = uuid.uuid4().hex
cache.set(token, student.id, timeout=60*60*24)
data = {
"msg": "登录成功",
"token": token
}
return jsonify(data)
else:
abort(400)
App/views/permission_book.py
知识点
Flask内置对象：g
from flask import Blueprint, request, g
from App.extension import cache
from App.models import Student
from App.models.sutdent_model import PERMISSION_STUDENT_LEARN, PERMISSION_STUDENT_MANAGE
book_blue = Blueprint("book_blue", __name__, url_prefix="/books")
@book_blue.route("/")
def books():
student = g.user
if not student.check_permission(PERMISSION_STUDENT_MANAGE):
return "没有访问本书的权限"
return "这是你喜欢的书<<算法导论>>，请拿走"
App/middleware.py
知识点：
面向切面编程
中间件
两个概念：切点和切面
切点：可以在那切开
切面：从切点处切开能获得什么
flask内置对象：g
g:可以跨函数使用，将数据传到App/views/permission_book.py
缓存
token
from flask import request, g
from App.extension import cache
from App.models import Student
require_login_list = ["/books/",]
def load_middleware(app):
@app.before_request
def before_request():
path = request.path
if path in require_login_list:
token = request.args.get("token")
if not token:
return "token 不存在"
student_id = cache.get(token)
if not student_id:
return "id 已过期"
student = Student.query.get(student_id)
if not student:
return "student 已消失"
g.user = student
g.auth = token
