项目地址：https://github.com/MasonYyp/myflask

1 安装python基础环境

# 安装flask
pip install flask# 安装redis
pip install redis# 安装操作MySQL的包
pip install flask-sqlalchemy
pip install pymysql
# 安装序列化包
pip install flask-marshmallow
pip install marshmallow-sqlalchemy# 安装生成token相关的包
pip install pyjwt# 安装生成验证码的包
pip install captcha

2 Docker安装redis和mysql

2.1 redis.conf

# redis.conf
# 可远程连接
# bind 127.0.0.1
# 解除保护模式
protected-mode no
# 数据持久化
appendonly yes
# 设置密码
requirepass 123456

2.2 创建redis容器

sudo docker run -itd \
--restart=always \
--name fl_redis \
-p 6379:6379 \
-v /home/fl_redis/redis.conf:/etc/redis/redis.conf \
-v /home/fl_redis/data:/data \
redis:6.2.2 redis-server /etc/redis/redis.conf

2.3 创建MySQL容器

# 安装MySQL，密码是123456
sudo docker run -itd \
--restart=always \
--name fl_mysql \
-p 3306:3306 \
-v /home/mysql/conf:/etc/mysql/conf.d \
-v /home/mysql/data:/var/lib/mysql \
-v /home/mysql/log:/var/log/mysql \
-e MYSQL_ROOT_PASSWORD=123456 \
mysql:5.7

进入MySQL容器创建数据库

create database myflask default character set utf8mb4 collate utf8mb4_unicode_ci;

3 vue前端

3.1 前端页面

<template><div class="login"><div>Login</div><div class="lo_cont"><div>用户名：<el-input v-model="name" placeholder="请输入用户名"></el-input></div><div>密码：<el-input v-model="pwd" placeholder="请输入密码" show-password></el-input></div><div>验证码：<img v-bind:src="codeImgSrc"><el-input v-model="code" placeholder="请输入验证码"></el-input></div><el-button v-on:click="login">提交</el-button></div></div>
</template><script>import axios from 'axios'// 按需引入ElementUI组件
import 'element-ui/lib/theme-chalk/index.css' 
import { Input, Button } from 'element-ui'export default {name: 'Login',props: {msg: String},components:{elInput: Input,elButton: Button},data:function(){return{name:"mason",pwd:"123456",code:"",codeKey: "",codeImgSrc: ""}},created:function(){this.get_captcha()},methods:{get_captcha: function(){axios.post("/flask/public/captcha").then(res=>{console.log(res.data)let data = res.data;this.codeImgSrc = data.code_img;this.codeKey = data.code_key;})},login: function(){let param = {name: this.name,pwd: this.pwd,code_key: this.codeKey,code: this.code}// 使用json方式传参axios.post("/flask/public/login", param).then(res=>{let token = res.dataconsole.log(token);})}}
}
</script><!-- Add "scoped" attribute to limit CSS to this component only -->
<style scoped>.lo_cont>>>.el-input{width: 30%;
}</style>

3.2 截图

前端截图

4 flask工程

4.1 工程目录

4.2 代码

4.2.1 main

from flask import Flaskfrom route.interceptor import before_interceptor
from route.operate_blueprint import OperateBlueprint
from dao.base_db.mysql_db import init_mysql_dbapp = Flask(__name__)# 设置拦截器
@app.before_request
def route_interceptor():return before_interceptor()# 初始化蓝本
operate_blueprint = OperateBlueprint(app)
operate_blueprint.init_blueprint()# 初始化MySQL
init_mysql_db(app)# Run the app
if __name__ == '__main__':app.run(host='0.0.0.0', port=5000)

4.2.2 config目录

4.2.2.1 common.py

import osdef get_root_path():cur_path = os.path.abspath(os.path.dirname(__file__))root_path, _ = os.path.split(cur_path)return root_path

4.2.2.2 db_conf.py

class RedisConf:# 配置基本参数pwd = "123456"host = "192.168.108.100"port = 6379class MysqlConf:acc = "root"pwd = "123456"host = "192.168.108.100"port = 3306db = "myflask"redis_conf = RedisConf()
mysql_conf = MysqlConf()

4.2.3 dao目录

4.2.3.1 base_db.mysql_db.py

from flask import Flask
from config.db_conf import mysql_conf
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow# 添加pymysql驱动，连接MySQL数据库
import pymysql
pymysql.install_as_MySQLdb()# 创建MySQL单实例
mysql_db = SQLAlchemy()# 创建Schema
mysql_schema = Marshmallow()# 初始化MySQL数据库
def init_mysql_db(app: Flask):# 配置MySQL数据库urldb_url = "mysql://" + mysql_conf.acc + ":" + mysql_conf.pwd + "@" + mysql_conf.host + ":" + str(mysql_conf.port) + "/" + mysql_conf.dbapp.config["SQLALCHEMY_DATABASE_URI"] = db_url# 关闭sqlalchemy自动跟踪数据库app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False# 显示底层执行的SQL语句app.config['SQLALCHEMY_ECHO'] = True# 解决‘No application found. Either work inside a view function or push an application context.’app.app_context().push()# 初始化appmysql_db.init_app(app)# 初始化schemamysql_schema.init_app(app)# 初始化table
def init_table():# 删除表mysql_db.drop_all()# 创建表mysql_db.create_all()

4.2.3.2 base_db.redis_db.py

import redis
from config.db_conf import redis_confclass RedisDB:# 初始化redisdef __init__(self):# 设置主机、端口号和密码redis_pool = redis.ConnectionPool(host=redis_conf.host, port=redis_conf.port, password=redis_conf.pwd, decode_responses=True)self.__strict_redis = redis.StrictRedis(connection_pool=redis_pool)# 在redis中添加键值，并设置过期时间def set(self, key, value, expiry):self.__strict_redis.set(name=key, value=value, ex=expiry)# 获取值def get(self, key):return self.__strict_redis.get(name=key)# 获取键值的剩余时间def ttl(self, key):# Time To Livereturn self.__strict_redis.ttl(name=key)# 设置单例模式
redis_db = RedisDB()

4.2.3.3 base_db.mysql_db.model.user_model.py

from dao.base_db.mysql_db import mysql_db as db# 创建用户表
class User(db.Model):# 用户__tablename__ = "tf_user"id = db.Column(db.Integer, nullable=False, primary_key=True, autoincrement=True)name = db.Column(db.String(32), nullable=False, unique=True)pwd = db.Column(db.String(32), nullable=False)def __repr__(self):# 显示对象中的信息return "User object: name=%s" % self.name

4.2.3.4 base_db.mysql_db.schema.user_schema.py

from dao.base_db.mysql_db import mysql_schema as ma
from dao.mysql_dao.model.user_model import User"""
# 序列化方法1
# 需要
注意flask-marshmallow的版本：flask-marshmallow<0.12.0class AuthorSchema(ma.ModelSchema)flask-marshmallow>=0.12.0 (recommended)class AuthorSchema(ma.SQLAlchemyAutoSchema)
"""# 使用flask_marshmallow初始化model
class UserSchema(ma.SQLAlchemyAutoSchema):class Meta:model = Userload_instance = True"""# 序列号方法2
from marshmallow import Schema, fields# 使用marshmallow初始化model
class UserSchema(Schema):id = fields.Integer()name = fields.String()pwd = fields.String()class Meta:# 设置序列化字段fields = ["id", "name", "pwd"]# 转化为有序字典ordered = True"""

4.2.3 route目录

4.2.3.1 interceptor.py

from flask import request
from dao.base_db.redis_db import redis_db# 设置拦截器
def before_interceptor():# Pass white listif white_list():passelse:res_status = check_token()if res_status == 1:passelse:return res_status# 设置白名单
def white_list():url_white_list = ["/flask/public/captcha", "/flask/public/login", "/flask/public/register", "/flask/public/initdb"]cur_url = request.path# Pass the urlif cur_url in url_white_list:return Trueelse:return False# 设置token
def check_token():token = request.headers.get("token")if token is None:return "Token None"token_ttl = redis_db.ttl(token)# 验证码已经过期if token_ttl <= 0:return "Token expired"# 获取redis中的coderedis_token = redis_db.get(token)if token != redis_token:return "Token error"return 1

4.2.3.2 operate_blueprint.py

# Configure Blueprintfrom route.public_route import public_bp
from route.userinfo import userinfo_bpclass OperateBlueprint():# Init the appdef __init__(self, app):self.__app = appself.__base_path = "/flask"# 初始化蓝本def init_blueprint(self):# Register the blueprintself.__app_register_blueprint(public_bp)self.__app_register_blueprint(userinfo_bp)# 在Flask中添加蓝本def __app_register_blueprint(self, blueprint):self.__app.register_blueprint(blueprint, url_prefix=self.__base_path + blueprint.name)

4.2.3.3 public_route.py

from flask import Blueprint, jsonify, request
from service.public_service import register_user, login_user, generate_captcha, create_tablepublic_bp = Blueprint("/public", __name__)@public_bp.route("/captcha", methods=['POST'])
def captcha():captcha_dict = generate_captcha()return jsonify(captcha_dict)@public_bp.route("/login", methods=['POST'])
def login():# 获取json数据data = request.get_json()print(data)res = login_user(data['name'], data['pwd'], data['code_key'], data['code'])return jsonify(res)@public_bp.route("/register", methods=['POST'])
def register():register_user()return "register"@public_bp.route("/initdb", methods=['POST'])
def init_mysql_db():return create_table()

4.2.3.4 userinfo.py

from flask import Blueprint
userinfo_bp = Blueprint("/userinfo", __name__)@userinfo_bp.route("/info", methods=['POST'])
def login():return "Mason"

4.2.4 service目录

public_service.py

from dao.base_db.redis_db import redis_db
from dao.base_db.mysql_db import mysql_db, init_tablefrom dao.mysql_dao.model.user_model import User
from dao.mysql_dao.schema.user_schema import UserSchemafrom util.operate_captcha import operate_captcha
from util.operate_token import operate_tokendef generate_captcha():# 生成验证码code = operate_captcha.generate_code()# 生成图片验证码image_base64_str = operate_captcha.generate_captcha_base64(code)# 生成唯一keycode_key = operate_captcha.generate_code_key()# 存入Redisredis_db.set(code_key, code.casefold(), 30)data = {"code_key": code_key,"code_img": image_base64_str}return datadef register_user():user = User()user.name = "mason"user.pwd = "123456"# 添加数据mysql_db.session.add(user)mysql_db.session.commit()def login_user(name, pwd, code_key, code):code_ttl = redis_db.ttl(code_key)# 验证码已经过期if code_ttl <= 0:return -1# 获取redis中的coderedis_code = redis_db.get(code_key)# 验证码过期if redis_code != code.casefold():return -2# 查询数据user = User.query.filter_by(name=name, pwd=pwd).first()# 序列号对象user_json = UserSchema().dump(user)print(user_json)# 生成tokentoken = operate_token.create_token(user.id, user.name, 60)# 添加到redisredis_db.set(token, token, 60)return tokendef create_table():init_table()return "create table"

4.2.5 util目录

4.2.5.1 operate_captcha.py

from captcha.image import ImageCaptcha
import randomimport base64
import uuidclass OperateCaptcha:def generate_code(self, code_len=4):code_chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'chars_len = len(code_chars) - 1code = ''for _ in range(code_len):index = random.randint(0, chars_len)code = code + code_chars[index]return codedef generate_captcha_base64(self, code):# 生成验证码image = ImageCaptcha()# 生成图片BytesIOimg_bytes_io = image.generate(code)# 转化为字符串image_base64 = base64.b64encode(img_bytes_io.read())image_base64_str = "data:image/png;base64," + str(image_base64, 'utf-8')return image_base64_strdef generate_code_key(self):# 生成唯一标识code_key = str(uuid.uuid1())return code_keyoperate_captcha = OperateCaptcha()

4.2.5.2 operate_token.py

import jwt
from datetime import datetime, timedelta# 操作Token
class OperateToken:def __init__(self):self._private_key = "123456"pass# 创建tokendef create_token(self, user_id, user_name, expiry_seconds):# Calculate expiry timeexpiry_time = datetime.utcnow() + timedelta(seconds=expiry_seconds)payload = {'exp': expiry_time,'user_id': user_id,'username': user_name}# 加密encode_jwt = jwt.encode(payload, self._private_key, algorithm='HS256')return encode_jwt# 验证tokendef check_token(self, token):decode_jwt = "-1"# 解密try:decode_jwt = jwt.decode(token, self._private_key, algorithms=['HS256'])except jwt.PyJWTError:print("Token is error!")return decode_jwtoperate_token = OperateToken()

5 postman截图

5.1 初始化数据表

http://127.0.0.1:5000/flask/public/initdb

5.2 注册接口

http://127.0.0.1:5000/flask/public/register

5.3 调用info接口

调用权限口需要验证token

http://127.0.0.1:5000/flask/userinfo/info