简介项目是一套基于 Python 与 Flask 框架、使用 MySQL 数据库进行持久化存储的情绪日记 APP 后端源码适合具备基础 Python 语法、想深入练习 Flask 路由、数据库读写与接口鉴权的初中级开发者也可作为毕业设计或小型系统后端的完整参考。压缩包共 45 个文件、约 3.99MB文件构成覆盖 14 个 .py 源文件、13 个 .pyc 编译文件、4 个 XML/IML 配置、TXT 文本、Git 忽略规则、SSL 证书与 License 授权说明源文件包含登录、日记管理、情绪评分等核心业务逻辑编译文件可直接被解释器加载配置与证书则帮助项目在真实环境中安全部署。工程按 Controller、Service、Pojo 分层并细化为 login、diary、emotion 等模块登录采用 JWT 校验日记模块支持增删改查computeScore.py 负责把用户描述转换为情绪分数test.py 可用于接口回归同时提供 .tsv 名人名言数据、说明文档和授权文件方便本地调试、二次开发与上线部署。已有 398 人学习对想将 FlaskMySQL 落到完整项目中的读者有较强参考价值。1. 情绪日记APP的后端起点Flask与MySQL到底怎么配合一个情绪日记APP最容易被低估的是后端那张“关系表”比前端界面更难设计。用户要写日记、打情绪分、按日期翻页、看周报统计落到底层全是这样的关系一个用户对应多条日记一条日记对应一个情绪标签同一天不能写两篇。用Flask加MySQL来搭这套后端比只用Python字典内存或NoSQL更合适因为日记和情绪统计天然需要join和group by。这个标题讲的是数据库操纵层面的设计不是UI不是部署脚本而是表怎么建、连接池怎么给、登录token怎么校验、并发写日记时锁会卡在哪。适合已经装好Python和MySQL、能写简单接口但还没把这三样组织成完整工程的开发者。2. 先把Mysql表结构和Flask项目骨架立起来2.1 MySQL表结构怎么拆用户、日记、情绪标签三张表2.1.1 为什么用户和日记不能合并成一张表情绪日记的最小模型是“某个用户在某天写了一段话选了一个情绪标签和一个强度分”。如果图省事把所有字段塞进一张表会出现两个问题一是用户名和密码哈希在每一条日记里重复存储密码一改就要扫全表二是以后要加“多种情绪标签”或者“情绪触发器”字段时只能不断加列整张表越来越像Excel。标准做法是拆成三张表users保存账号mood_tags保存预置情绪类型diary_entries保存日记正文和评分。这样外键关系清晰统计“每种情绪出现多少次”只需要group by一张小表。CREATE TABLE users ( id INT PRIMARY KEY AUTO_INCREMENT, username VARCHAR(64) NOT NULL UNIQUE, password_hash VARCHAR(255) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_unicode_ci; CREATE TABLE mood_tags ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(32) NOT NULL UNIQUE, color VARCHAR(16) NOT NULL DEFAULT #8e44ad, sort_order INT NOT NULL DEFAULT 0 ) ENGINEInnoDB; CREATE TABLE diary_entries ( id BIGINT PRIMARY KEY AUTO_INCREMENT, user_id INT NOT NULL, mood_tag_id INT, content TEXT NOT NULL, mood_score TINYINT NOT NULL COMMENT 情感强度分 1-10, entry_date DATE NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, CONSTRAINT fk_entries_user FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE, CONSTRAINT fk_entries_tag FOREIGN KEY (mood_tag_id) REFERENCES mood_tags(id) ON DELETE SET NULL, UNIQUE KEY uk_user_date (user_id, entry_date) ) ENGINEInnoDB;这段DDL最大的价值在三个地方engineInnoDB保证事务和外键可用entry_date用DATE类型跟created_at分开业务日期和服务器写入时间去耦uk_user_date唯一索引从数据库层面兜底“同一天只能写一篇”的约束哪怕应用层漏判一次重复提交MySQL也会挡住第二次插入。2.1.2 字段配置和字符集的关键参数字段/参数取值为什么要这么配contentTEXT日记可能超过255字用VARCHAR(255)会截断或报错mood_scoreTINYINT只存1-10TINYINT已够还带COMMENT方便DBA识别utf8mb4库表三级移动端输入法会写emojiutf8只有3字节会入库失败uk_user_date唯一索引业务上一天只能一条防止前端双击导致重复日记实际项目中我曾经只把content定为VARCHAR(255)结果一个用户在手机备忘录里写长文粘进来插入直接报Data too long。后来切成TEXT解决。字符集这里建议不要在运行后再改建库时就用CREATE DATABASE mood_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;避免后续迁移数据时踩Incorrect string value。2.2 Flask工厂模式与配置文件拆分2.2.1 为什么用工厂模式而不是全局appFlask单文件也能跑但情绪日记后端要写接口测试、要区分开发和生产配置再用全局app Flask(__name__)就会越写越乱。工厂模式把创建过程包成一个函数测试时可以用独立的配置实例化另一个app不污染开发环境。# config.py class Config: SQLALCHEMY_DATABASE_URI mysqlpymysql://mood_app:your_password127.0.0.1:3306/mood_db?charsetutf8mb4 SQLALCHEMY_TRACK_MODIFICATIONS False SQLALCHEMY_ENGINE_OPTIONS { pool_size: 10, pool_recycle: 3600, pool_pre_ping: True, } SECRET_KEY set-a-random-value-in-env2.2.2 创建应用并注册蓝图# app/__init__.py from flask import Flask from flask_sqlalchemy import SQLAlchemy from config import Config db SQLAlchemy() def create_app(config_classConfig): app Flask(__name__) app.config.from_object(config_class) db.init_app(app) with app.app_context(): from .api import api_bp app.register_blueprint(api_bp) return app这段代码里容易看漏的是连接串末尾的?charsetutf8mb4。它必须挂在query string里否则PyMySQL默认用latin1连MySQL中文写入直接乱码。另一个细节是db.init_app(app)放在from_object之后这样Config里的SQLALCHEMY_DATABASE_URI能立刻被读取。2.3 MySQL连接池参数要调的三个点2.3.1 短连接导致握手开销Flask每次请求结束会执行db.session.remove()但底层MySQL连接并不是立即关闭而是归还给连接池。如果不调整连接池SQLAlchemy默认pool_size是5一个情绪日记应用刚上线没多少人用没问题一旦出现早上九点集中打卡写心情的场景连接队列会开始等待。参数默认值建议值配置位置pool_size510~20SQLALCHEMY_ENGINE_OPTIONSpool_recycle36003600同上pool_pre_pingFalseTrue同上2.3.2 pool_pre_ping 为什么必须 TrueMySQL的wait_timeout默认8小时连接池里的连接如果一直空闲超过这个时间再取出来用会报MySQL server has gone away。pool_pre_pingTrue会在从池里取连接时先发一条SELECT 1探测省得请求打到一半才发现连接断了。pool_recycle要小于数据库侧的wait_timeout如果DBA把超时时间调成了4小时你就得把pool_recycle改成3000留10%余量让连接在过期前主动重建。SQLALCHEMY_ENGINE_OPTIONS { pool_size: 10, pool_recycle: 3000, pool_pre_ping: True, }这个写法和默认值的差别在于pool_recycle是“连接存在多久后强制丢弃”pool_pre_ping是“每次取连接时先验活”。两个都开才能同时解决空闲超时和网络中断导致的连接失效。刚开始做Flask后端的人常常只调pool_size等线上出现偶发的OperationalError时才想起来补后两个参数。3. 操纵Mysql数据的核心ORM模型、迁移与CRUD服务层3.1 用SQLAlchemy映射MySQL表情绪字段这样定义直接用原生的pymysql写SQL不是不行但Flask项目里视图函数越多手写字符串拼接的查重和维护成本越高。SQLAlchemy的ORM层让我们用Python对象操作MySQL同时还能保留执行原生SQL的能力。# app/models.py from datetime import datetime from . import db class User(db.Model): __tablename__ users id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(64), uniqueTrue, nullableFalse) password_hash db.Column(db.String(255), nullableFalse) created_at db.Column(db.DateTime, defaultdatetime.utcnow) class MoodTag(db.Model): __tablename__ mood_tags id db.Column(db.Integer, primary_keyTrue) name db.Column(db.String(32), uniqueTrue, nullableFalse) color db.Column(db.String(16), default#8e44ad) sort_order db.Column(db.Integer, default0) class DiaryEntry(db.Model): __tablename__ diary_entries id db.Column(db.BigInteger, primary_keyTrue) user_id db.Column(db.Integer, db.ForeignKey(users.id), nullableFalse) mood_tag_id db.Column(db.Integer, db.ForeignKey(mood_tags.id)) content db.Column(db.Text, nullableFalse) mood_score db.Column(db.SmallInteger, nullableFalse) entry_date db.Column(db.Date, nullableFalse) created_at db.Column(db.DateTime, defaultdatetime.utcnow)这里要注意MySQL表里的TINYINT映射成SQLAlchemy的SmallInteger不要直接用Integer否则自动生成的迁移脚本会把列改成int(11)虽然不影响功能但在后续用flask-migrate做diff时会看到一堆无意义的变更。ORM模型里的__tablename__必须和第二章的DDL对齐如果不一样db.create_all()会再建一张新表到时候排查半天也不知道为什么表多出来了。3.2 注册登录接口里的密码哈希与JWT签发参数3.2.1 不要明文存密码用Werkzeug哈希Flask生态里Werkzeug自带的generate_password_hash是首选它把随机盐和哈希算法一起编码进结果字符串里不需要额外建盐字段。注册接口的核心代码# app/services/auth.py from werkzeug.security import generate_password_hash, check_password_hash from app.models import User from app import db def create_user(username, password): if User.query.filter_by(usernameusername).first(): raise ValueError(username already exists) user User(usernameusername, password_hashgenerate_password_hash(password)) db.session.add(user) db.session.commit() return user.id参数说明generate_password_hash(password, methodpbkdf2:sha256, salt_length16)这一行如果不显式传method会使用Werkzeug版本默认算法。不同版本之间默认值可能从pbkdf2:sha1换成pbkdf2:sha256导致老密文校验失败。建议把method和salt_length写死在代码里升级依赖时不会隐性改变行为。3.2.2 JWT过期时间和密钥配置账号密码校验通过后签发一个带过期时间的token。用PyJWT库来做import jwt from datetime import datetime, timedelta from flask import current_app def issue_token(user_id): expire datetime.utcnow() timedelta(hourscurrent_app.config[JWT_EXPIRE_HOURS]) payload { uid: user_id, exp: expire, } return jwt.encode(payload, keycurrent_app.config[SECRET_KEY], algorithmHS256)JWT_EXPIRE_HOURS这个参数建议在Config里设成24移动端用户一天打开一次日记应用token有效期太长有被盗用风险太短又要频繁重新登录。SECRET_KEY必须从环境变量读取不能写死在config.py里提交进Git仓库。3.3 日记列表查询减少循环查询的分页参数3.3.1 joinedload加载避免N1次数据库往返列表接口最典型的性能问题是先查日记再逐条查对应的mood_tag。一页20条就是21次MySQL往返网络延迟稍微高一点接口响应时间直接翻倍。用joinedload让SQLAlchemy生成一条LEFT JOIN把标签一起查出来# app/routes.py from flask import Blueprint, request, jsonify from sqlalchemy.orm import joinedload from app.models import DiaryEntry from app.auth import auth_required api_bp Blueprint(api, __name__, url_prefix/api) api_bp.get(/entries) def list_entries(): uid auth_required() page int(request.args.get(page, 1)) per_page min(int(request.args.get(limit, 20)), 100) query DiaryEntry.query.options( joinedload(DiaryEntry.mood_tag) ).filter(DiaryEntry.user_id uid) rows query.order_by(DiaryEntry.entry_date.desc()).paginate( pagepage, per_pageper_page, error_outFalse ) return jsonify({ total: rows.total, items: [serialize(r) for r in rows.items] })joinedload只对当前这次查询生效不会改变模型默认加载方式。它生成的SQL是先LEFT JOIN mood_tags再ORDER BY entry_date DESC如果entry_date上没有索引MySQL会先做一次filesort。所以下面这个复合索引不能漏参数/索引默认情况建议配置说明page11超出总页数时返回空数组不报404limit201~100超过100强制截断防止整表拉取索引只有主键(user_id, entry_date)覆盖where和order by两个条件3.3.2 分页参数serro_outFalse的边界行为paginate默认error_outTrue一旦请求页码超过总页数就返回404。用户在最后一页删除了一条日记后翻页前端会收到404然后弹错误提示体验很差。所以这里显式传error_outFalse让它返回空列表前端根据total判断是否还有下一页。per_page的截断逻辑要在paginate之前做因为min()只限制上限如果传limit0SQLAlchemy会认为是要获取所有数据所以再加一层max(1, ...)更稳。4. 情绪日记API的Flask视图与MySQL事务参数4.1 请求上下文中的用户身份从Authorization头到数据库查询4.1.1 视图函数里如何拿到当前用户登录之后的每一个日记操作都要知道当前用户是谁不能相信前端传来的user_id。标准做法是从Authorization: Bearer token头里解出uid再去查数据库。PyJWT从某个版本开始要求必须显式传algorithms参数否则直接抛异常def auth_required(): auth_header request.headers.get(Authorization, ) if not auth_header.startswith(Bearer ): abort(401, descriptionmissing bearer token) token auth_header[7:] try: payload jwt.decode( token, keycurrent_app.config[SECRET_KEY], algorithms[HS256] ) except jwt.ExpiredSignatureError: abort(401, descriptiontoken expired) except jwt.InvalidTokenError: abort(401, descriptioninvalid token) return payload[uid]异常类型场景返回给前端的提示ExpiredSignatureErrortoken过期时间到了token expired移动端跳登录页DecodeErrortoken被篡改或不是合法JWTinvalid tokenInvalidAlgorithmError后端又换了签名算法需要让客户端重新获取token这里有个细节token auth_header[7:]是硬编码猜Bearer恰好是6个字符加一个空格。为了更严谨可以用auth_header.split( )取第二个元素因为如果前端拼了多个空格我的做法会取到空字符串而split()方式会容错一些。选择这种写法只是因为它短生产代码我一般还会检查len(parts) 2。4.2 MySQL时区字段的坑CURRENT_TIMESTAMP与created_at4.2.1 按业务日期用entry_date不依赖created_at情绪日记的关键查询是“某一天写了什么”这个日期应该由客户端在提交正文时一并传过来而不是服务器收到请求的时间。created_at只用于审计和排序。如果把统计逻辑建立在created_at上用户跨时区旅行时日记日期会跳到前一天或后一天周报就会少一条数据。from datetime import date data request.get_json() entry_date date.fromisoformat(data[entry_date]) # 客户端传 2025-01-15date.fromisoformat是Python 3.7才有的方法如果项目还在跑Python 3.6这里会直接报AttributeError。建议在项目的runtime.txt或Dockerfile里固定Python 3.10省去低版本兼容补丁。客户端传日期时只传日期不要带时区后缀比如2025-01-15T00:00:0008:00否则fromisoformat在Python 3.10里能解析3.7会报格式错误。4.2.2 MySQL时区变量如何和连接串配合如果数据库服务器时区不是UTC而应用服务器是UTC那么DEFAULT CURRENT_TIMESTAMP写入的是数据库当前时区跟Python的datetime.utcnow()存进去的值会差几个小时。统一办法是把MySQL时区设成UTC所有API只接受明确的entry_date前端要显示本地时间就自己按用户时区偏移。生产环境可以在MySQL的配置里加[mysqld] default-time-zone 00:00如果改动数据库配置要重启临时想验证一个会话可以执行SET GLOBAL time_zone 00:00;但:g表示这个重启后还在需要确保写入配置文件。注意不要相信移动端设备时间戳能替代后端日期字段。用户可能主动改手机时间entry_date应该由服务端在创建日记时做一次合法性校验范围不超过当天前后1天以此阻断脏数据进入统计表。4.3 并发写日记时的MySQL行锁与死锁处理4.3.1 用唯一键约束兜底并发正常情况下用户一天只写一条日记。如果两个请求同时到达应用层都执行SELECT发现不存在然后两个都插入最终只有一个成功另一个被IntegrityError捕获并返回409这就是uk_user_date唯一索引在做幕后兜底。Flask视图里要捕获SQLAlchemyError并rollbackfrom sqlalchemy.exc import IntegrityError try: db.session.add(entry) db.session.commit() except IntegrityError: db.session.rollback() abort(409, descriptionentry already exists for this date)这里不需要手动调整事务隔离级别MySQL默认的REPEATABLE_READ配合唯一索引已经能完成插入去重。唯一键冲突的错误捕获范围要控制好别把Not Null约束的错误也当成409返回。4.3.2 修改日记内容时用 FOR UPDATE 锁行编辑日记的接口如果只是先查再改两个请求同时来后者会覆盖前者内容。更稳的做法是查询时加with_for_update()让事务在提交前锁住这一行entry DiaryEntry.query.filter_by( identry_id, user_iduid ).with_for_update().first_or_404() try: entry.content request.json[content] db.session.commit() except OperationalError as exc: db.session.rollback() if Deadlock in str(exc): abort(409, descriptiondeadlock, retry later) raisewith_for_update()生成的SQL是SELECT ... FOR UPDATE在事务提交前其他事务无法更新这一行。死锁最容易发生在两个请求同时更新多条日记但顺序不一致的情况。举例请求A先锁日记id5再锁id3请求B先锁id3再锁id5两边就会互相等待MySQL检测到死锁后直接把其中一个事务回滚。避免办法是每次操作多行时按主键升序锁或者让前端编辑时带上updated_at做乐观锁冲突时返回409并让用户重试。并发动作表和锁对象冲突时的表现推荐处理同一天插日记uk_user_date索引第二个insert收到重复键返回409并拉取最新数据同一条日记编辑主键行锁后一个事务等待锁超时前端做乐观锁带上updated_at条件删除情绪标签mood_tags行 diary_entries外键子表外键阻塞先批量SET NULL再删除5. Flask接口上线前要调的MySQL参数与验证方法5.1 三个必须调整的数据库服务端参数Flask开发时用默认MySQL配置够用但线上部署前有几个参数必须改。max_connections默认151如果你用gunicorn起了4个worker每个worker连接池10那连接池总和已经40再加上后台任务和管理工具离上限不远。innodb_buffer_pool_size默认128M日记表一旦积累到几万行统计查询会频繁读磁盘。参数默认值推荐值原因max_connections151300或按连接池总和*2防止Flask多进程下连接池总和超过上限innodb_buffer_pool_size128M物理内存60%日记表读取频繁热点数据尽量留在内存innodb_lock_wait_timeout503~10快速暴露业务死锁而不是让请求一直卡住修改示例# /etc/mysql/mysql.conf.d/mysqld.cnf [mysqld] max_connections 300 innodb_buffer_pool_size 1G innodb_lock_wait_timeout 5改完重启MySQL后用SHOW VARIABLES LIKE innodb_lock_wait_timeout;确认参数已生效不要凭印象觉得改了配置文件就有用。5.2 用curl验证Flask接口的headers和数据库索引命中的方法启动服务后先看认证头是否被正确读取。这里有一个技巧不要用那种只打API的工具直接看原始响应头能确认Flask有没有返回预期状态码和CORS头。curl -i -H Authorization: Bearer $TOKEN \ http://127.0.0.1:5000/api/entries?page1limit20 \ -s | head -20然后打开MySQL慢查询日志或者直接用EXPLAIN验证分页查询是否走索引EXPLAIN SELECT * FROM diary_entries WHERE user_id 1 ORDER BY entry_date DESC LIMIT 20;重点看type列和key列。如果看到typeALL全表扫描就补上复合索引ALTER TABLE diary_entries ADD INDEX idx_user_date (user_id, entry_date);加完之后再用一次EXPLAINkey那列应该变成idx_user_datetype至少是ref。线上压测阶段观察SHOW STATUS LIKE Innodb_rows_read增长量如果QPS上去了但Innodb_rows_read不再暴涨说明索引把大部分查询落在了索引树里而不是一行行扫磁盘。这套验证做完再谈调用第三方情绪分析接口也不迟。本文还有配套的精品资源点击获取