Python之Flask开发框架(第四篇)

经过前三篇的系统学习，我们已经掌握了 Flask 从入门到进阶的完整知识体系。本篇将对 Flask 的核心知识进行总结，并在此基础上提供一个完整的、可直接运行的简易博客系统，将所学内容融会贯通，帮助你巩固知识并快速应用于实际项目。

一、Flask 知识体系回顾

1. 核心概念

• 应用（Application）
  ：Flask 类的实例，是整个应用的入口。
• 路由（Route）
  ：使用 @app.route 装饰器将 URL 映射到视图函数，支持动态 URL 和多种 HTTP 方法。
• 视图函数（View Function）
  ：处理请求并返回响应，支持返回字符串、字典、元组、Response 对象等。
• 请求（Request）
  ：request 对象封装了客户端发送的请求信息，如表单数据、查询参数、JSON 数据、请求头等。
• 响应（Response）
  ：视图函数的返回值会被 Flask 自动转换为 Response 对象，可以手动构建更复杂的响应。
• 会话（Session）
  ：基于签名 cookie 的会话管理，用于跨请求保存用户数据。

2. 模板与静态文件

• Jinja2 模板引擎
  ：支持变量、控制结构、过滤器、宏、模板继承。
• 静态文件
  ：存放在 static 目录，通过 url_for('static', filename='...') 生成 URL。

3. 表单处理

• Flask-WTF
   扩展集成 WTForms，提供 CSRF 保护、表单验证、字段渲染等功能。
• 表单类
  ：定义字段和验证器，在视图函数中实例化并验证。

4. 数据库操作

• Flask-SQLAlchemy
   提供 ORM 能力，简化数据库操作。
• 模型定义
  ：继承 db.Model，使用 db.Column 定义字段，支持关系映射（一对多、多对多）。
• 迁移管理
  ：通过 Flask-Migrate 自动化生成迁移脚本，管理数据库版本。

5. 蓝图（Blueprint）

• 模块化
  ：将相关路由、视图、模板等组织成蓝图，通过 app.register_blueprint 注册到应用。
• URL 前缀
  ：在蓝图创建时指定 url_prefix，实现路由分组。

6. 请求钩子与中间件

• before_request、after_request、teardown_request
  ：在请求生命周期各阶段执行通用逻辑。
• WSGI 中间件
  ：将 Flask 应用包装在 WSGI 中间件中，扩展功能（如性能分析、代理处理）。

7. 常用扩展

8. 生产部署

• WSGI 服务器
  ：Gunicorn、uWSGI 等。
• 反向代理
  ：Nginx 处理静态文件、负载均衡、HTTPS 等。
• 进程管理
  ：Supervisor、systemd 等保证应用持续运行。
• 容器化
  ：Docker 打包应用与环境，便于部署和扩展。
• 云平台
  ：Heroku、PythonAnywhere、AWS 等。

二、完整简易博客系统

下面我们将前面的知识点整合成一个功能完备的博客系统。该系统具有以下特性：

• 用户注册、登录、登出（使用 Flask-Login）
• 注册时发送欢迎邮件（使用 Flask-Mail）
• 文章创建、编辑、删除（仅作者本人）
• 文章列表分页显示
• 首页缓存（使用 Flask-Caching）
• 使用蓝图组织路由
• 使用 SQLite 数据库（可轻松切换至 MySQL/PostgreSQL）
• 生产配置与环境变量管理

1. 项目结构

blog/
├── app/
│   ├── __init__.py
│   ├── models.py
│   ├── forms.py
│   ├── auth/
│   │   ├── __init__.py
│   │   └── routes.py
│   ├── main/
│   │   ├── __init__.py
│   │   └── routes.py
│   ├── templates/
│   │   ├── base.html
│   │   ├── auth/
│   │   │   ├── login.html
│   │   │   └── register.html
│   │   └── main/
│   │       ├── index.html
│   │       ├── post_detail.html
│   │       ├── create_post.html
│   │       └── edit_post.html
│   └── static/
│       └── style.css
├── config.py
├── run.py
├── requirements.txt
└── .env

2. 环境准备

创建虚拟环境并安装依赖：

python -m venv venv
source venv/bin/activate   # Windows: venv\Scripts\activate
pip install flask flask-sqlalchemy flask-migrate flask-wtf flask-login flask-mail flask-caching python-dotenv

requirements.txt 内容：

flask==2.3.3
flask-sqlalchemy==3.1.1
flask-migrate==4.0.5
flask-wtf==1.1.1
flask-login==0.6.2
flask-mail==0.9.1
flask-caching==2.1.0
python-dotenv==1.0.0

3. 配置文件 config.py

import os
from dotenv import load_dotenv

load_dotenv()

classConfig:
    SECRET_KEY = os.environ.get('SECRET_KEY')or'hard-to-guess-string'
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')or'sqlite:///blog.db'
    SQLALCHEMY_TRACK_MODIFICATIONS =False

# Flask-Mail 配置
    MAIL_SERVER = os.environ.get('MAIL_SERVER','smtp.qq.com')
    MAIL_PORT =int(os.environ.get('MAIL_PORT',587))
    MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS','true').lower()in['true','on','1']
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')

# Flask-Caching 配置
    CACHE_TYPE = os.environ.get('CACHE_TYPE','SimpleCache')

4. 应用工厂 app/__init__.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_mail import Mail
from flask_caching import Cache
from config import Config

db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
mail = Mail()
cache = Cache()

defcreate_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)

    db.init_app(app)
    migrate.init_app(app, db)
    login_manager.init_app(app)
    mail.init_app(app)
    cache.init_app(app)

    login_manager.login_view ='auth.login'
    login_manager.login_message ='请先登录'

from app.auth import auth_bp
from app.main import main_bp
    app.register_blueprint(auth_bp)
    app.register_blueprint(main_bp)

return app

5. 数据库模型 app/models.py

from app import db
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime

classUser(UserMixin, db.Model):
id= db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(128))
    posts = db.relationship('Post', backref='author', lazy='dynamic')

defset_password(self, password):
        self.password_hash = generate_password_hash(password)

defcheck_password(self, password):
return check_password_hash(self.password_hash, password)

def__repr__(self):
returnf'<User {self.username}>'

classPost(db.Model):
id= db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    body = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))

def__repr__(self):
returnf'<Post {self.title}>'

6. 表单类 app/forms.py

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField, TextAreaField
from wtforms.validators import DataRequired, Email, Length, EqualTo

classRegistrationForm(FlaskForm):
    username = StringField('用户名', validators=[DataRequired(), Length(min=2,max=64)])
    email = StringField('邮箱', validators=[DataRequired(), Email()])
    password = PasswordField('密码', validators=[DataRequired(), Length(min=6)])
    confirm = PasswordField('确认密码', validators=[DataRequired(), EqualTo('password')])
    submit = SubmitField('注册')

classLoginForm(FlaskForm):
    email = StringField('邮箱', validators=[DataRequired(), Email()])
    password = PasswordField('密码', validators=[DataRequired()])
    submit = SubmitField('登录')

classPostForm(FlaskForm):
    title = StringField('标题', validators=[DataRequired()])
    body = TextAreaField('内容', validators=[DataRequired()])
    submit = SubmitField('发布')

7. 认证蓝图 app/auth/__init__.py

from flask import Blueprint

auth_bp = Blueprint('auth', __name__, url_prefix='/auth')

from app.auth import routes

8. 认证路由 app/auth/routes.py

from flask import render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, current_user
from app import db, mail
from app.auth import auth_bp
from app.forms import RegistrationForm, LoginForm
from app.models import User
from flask_mail import Message

@auth_bp.route('/register', methods=['GET','POST'])
defregister():
if current_user.is_authenticated:
return redirect(url_for('main.index'))
    form = RegistrationForm()
if form.validate_on_submit():
# 检查用户是否存在
        user = User.query.filter_by(username=form.username.data).first()
if user:
            flash('用户名已存在','danger')
return redirect(url_for('auth.register'))
        user = User.query.filter_by(email=form.email.data).first()
if user:
            flash('邮箱已被注册','danger')
return redirect(url_for('auth.register'))

        user = User(username=form.username.data, email=form.email.data)
        user.set_password(form.password.data)
        db.session.add(user)
        db.session.commit()

# 发送欢迎邮件
try:
            msg = Message('欢迎注册博客', recipients=[user.email])
            msg.body =f'你好 {user.username}，感谢注册我们的博客系统！'
            mail.send(msg)
except Exception as e:
            app.logger.error(f'邮件发送失败: {e}')

        flash('注册成功，请登录','success')
return redirect(url_for('auth.login'))
return render_template('auth/register.html', form=form)

@auth_bp.route('/login', methods=['GET','POST'])
deflogin():
if current_user.is_authenticated:
return redirect(url_for('main.index'))
    form = LoginForm()
if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
if user and user.check_password(form.password.data):
            login_user(user)
            next_page = request.args.get('next')
return redirect(next_page)if next_page else redirect(url_for('main.index'))
else:
            flash('邮箱或密码错误','danger')
return render_template('auth/login.html', form=form)

@auth_bp.route('/logout')
deflogout():
    logout_user()
return redirect(url_for('main.index'))

9. 主蓝图 app/main/__init__.py

from flask import Blueprint

main_bp = Blueprint('main', __name__)

from app.main import routes

10. 主路由 app/main/routes.py

from flask import render_template, redirect, url_for, flash, abort, request
from flask_login import login_required, current_user
from app import db, cache
from app.main import main_bp
from app.forms import PostForm
from app.models import Post

@main_bp.route('/')
@cache.cached(timeout=60)
defindex():
    page = request.args.get('page',1,type=int)
    posts = Post.query.order_by(Post.created_at.desc()).paginate(page=page, per_page=10)
return render_template('main/index.html', posts=posts)

@main_bp.route('/post/<int:id>')
defpost_detail(id):
    post = Post.query.get_or_404(id)
return render_template('main/post_detail.html', post=post)

@main_bp.route('/create', methods=['GET','POST'])
@login_required
defcreate_post():
    form = PostForm()
if form.validate_on_submit():
        post = Post(title=form.title.data, body=form.body.data, author=current_user)
        db.session.add(post)
        db.session.commit()
        flash('文章发布成功','success')
return redirect(url_for('main.index'))
return render_template('main/create_post.html', form=form)

@main_bp.route('/edit/<int:id>', methods=['GET','POST'])
@login_required
defedit_post(id):
    post = Post.query.get_or_404(id)
if post.author != current_user:
        abort(403)
    form = PostForm()
if form.validate_on_submit():
        post.title = form.title.data
        post.body = form.body.data
        db.session.commit()
        flash('文章已更新','success')
return redirect(url_for('main.post_detail',id=post.id))
elif request.method =='GET':
        form.title.data = post.title
        form.body.data = post.body
return render_template('main/edit_post.html', form=form, post=post)

@main_bp.route('/delete/<int:id>', methods=['POST'])
@login_required
defdelete_post(id):
    post = Post.query.get_or_404(id)
if post.author != current_user:
        abort(403)
    db.session.delete(post)
    db.session.commit()
    flash('文章已删除','success')
return redirect(url_for('main.index'))

11. 用户加载回调（在 app/__init__.py 中补充）

from app.models import User

@login_manager.user_loader
defload_user(user_id):
return User.query.get(int(user_id))

将这段代码放在 create_app 函数中，login_manager.init_app(app) 之后。

12. 模板文件

templates/base.html

<!DOCTYPEhtml>
<htmllang="zh">
<head>
<metacharset="UTF-8">
<metaname="viewport"content="width=device-width, initial-scale=1.0">
<title>{% block title %}博客系统{% endblock %}</title>
<linkrel="stylesheet"href="{{ url_for('static', filename='style.css') }}">
</head>
<body>
<nav>
<divclass="container">
<ahref="{{ url_for('main.index') }}"class="brand">我的博客</a>
<ulclass="nav-links">
                {% if current_user.is_authenticated %}
<li>你好，{{ current_user.username }}</li>
<li><ahref="{{ url_for('main.create_post') }}">写文章</a></li>
<li><ahref="{{ url_for('auth.logout') }}">登出</a></li>
                {% else %}
<li><ahref="{{ url_for('auth.login') }}">登录</a></li>
<li><ahref="{{ url_for('auth.register') }}">注册</a></li>
                {% endif %}
</ul>
</div>
</nav>
<mainclass="container">
        {% with messages = get_flashed_messages(with_categories=true) %}
            {% if messages %}
                {% for category, message in messages %}
<divclass="alert alert-{{ category }}">{{ message }}</div>
                {% endfor %}
            {% endif %}
        {% endwith %}
        {% block content %}{% endblock %}
</main>
<footer>
<divclass="container">
<p>&copy; 2023 我的博客</p>
</div>
</footer>
</body>
</html>

templates/auth/register.html

{% extends "base.html" %}

{% block title %}注册{% endblock %}

{% block content %}
<h2>注册</h2>
<formmethod="POST">
    {{ form.hidden_tag() }}
<divclass="form-group">
        {{ form.username.label }}<br>
        {{ form.username(size=32) }}<br>
        {% for error in form.username.errors %}
<spanclass="error">{{ error }}</span>
        {% endfor %}
</div>
<divclass="form-group">
        {{ form.email.label }}<br>
        {{ form.email(size=32) }}<br>
        {% for error in form.email.errors %}
<spanclass="error">{{ error }}</span>
        {% endfor %}
</div>
<divclass="form-group">
        {{ form.password.label }}<br>
        {{ form.password(size=32) }}<br>
        {% for error in form.password.errors %}
<spanclass="error">{{ error }}</span>
        {% endfor %}
</div>
<divclass="form-group">
        {{ form.confirm.label }}<br>
        {{ form.confirm(size=32) }}<br>
        {% for error in form.confirm.errors %}
<spanclass="error">{{ error }}</span>
        {% endfor %}
</div>
<divclass="form-group">
        {{ form.submit() }}
</div>
</form>
{% endblock %}

templates/auth/login.html

{% extends "base.html" %}

{% block title %}登录{% endblock %}

{% block content %}
<h2>登录</h2>
<formmethod="POST">
    {{ form.hidden_tag() }}
<divclass="form-group">
        {{ form.email.label }}<br>
        {{ form.email(size=32) }}<br>
        {% for error in form.email.errors %}
<spanclass="error">{{ error }}</span>
        {% endfor %}
</div>
<divclass="form-group">
        {{ form.password.label }}<br>
        {{ form.password(size=32) }}<br>
        {% for error in form.password.errors %}
<spanclass="error">{{ error }}</span>
        {% endfor %}
</div>
<divclass="form-group">
        {{ form.submit() }}
</div>
</form>
{% endblock %}

templates/main/index.html

{% extends "base.html" %}

{% block title %}首页{% endblock %}

{% block content %}
<h1>最新文章</h1>
{% for post in posts.items %}
<articleclass="post">
<h2><ahref="{{ url_for('main.post_detail', id=post.id) }}">{{ post.title }}</a></h2>
<divclass="post-meta">
            作者：{{ post.author.username }} | 发布于：{{ post.created_at.strftime('%Y-%m-%d') }}
</div>
<divclass="post-summary">
            {{ post.body[:200] }}{% if post.body|length > 200 %}...{% endif %}
</div>
        {% if current_user == post.author %}
<divclass="post-actions">
<ahref="{{ url_for('main.edit_post', id=post.id) }}">编辑</a>
<formaction="{{ url_for('main.delete_post', id=post.id) }}"method="POST"style="display:inline;">
<buttontype="submit"onclick="returnconfirm('确定删除吗？')">删除</button>
</form>
</div>
        {% endif %}
</article>
{% else %}
<p>暂无文章</p>
{% endfor %}

<divclass="pagination">
    {% if posts.has_prev %}
<ahref="{{ url_for('main.index', page=posts.prev_num) }}">上一页</a>
    {% endif %}
<span>第 {{ posts.page }} 页 / 共 {{ posts.pages }} 页</span>
    {% if posts.has_next %}
<ahref="{{ url_for('main.index', page=posts.next_num) }}">下一页</a>
    {% endif %}
</div>
{% endblock %}

templates/main/post_detail.html

{% extends "base.html" %}

{% block title %}{{ post.title }}{% endblock %}

{% block content %}
<articleclass="post-detail">
<h1>{{ post.title }}</h1>
<divclass="post-meta">
        作者：{{ post.author.username }} | 发布于：{{ post.created_at.strftime('%Y-%m-%d %H:%M') }}
</div>
<divclass="post-body">
        {{ post.body|safe }}
</div>
    {% if current_user == post.author %}
<divclass="post-actions">
<ahref="{{ url_for('main.edit_post', id=post.id) }}">编辑</a>
<formaction="{{ url_for('main.delete_post', id=post.id) }}"method="POST"style="display:inline;">
<buttontype="submit"onclick="returnconfirm('确定删除吗？')">删除</button>
</form>
</div>
    {% endif %}
</article>
{% endblock %}

templates/main/create_post.html

{% extends "base.html" %}

{% block title %}写文章{% endblock %}

{% block content %}
<h2>发布新文章</h2>
<formmethod="POST">
    {{ form.hidden_tag() }}
<divclass="form-group">
        {{ form.title.label }}<br>
        {{ form.title(size=64) }}<br>
        {% for error in form.title.errors %}
<spanclass="error">{{ error }}</span>
        {% endfor %}
</div>
<divclass="form-group">
        {{ form.body.label }}<br>
        {{ form.body(rows=15, cols=80) }}<br>
        {% for error in form.body.errors %}
<spanclass="error">{{ error }}</span>
        {% endfor %}
</div>
<divclass="form-group">
        {{ form.submit() }}
</div>
</form>
{% endblock %}

templates/main/edit_post.html

{% extends "base.html" %}

{% block title %}编辑文章{% endblock %}

{% block content %}
<h2>编辑文章</h2>
<formmethod="POST">
    {{ form.hidden_tag() }}
<divclass="form-group">
        {{ form.title.label }}<br>
        {{ form.title(size=64) }}<br>
        {% for error in form.title.errors %}
<spanclass="error">{{ error }}</span>
        {% endfor %}
</div>
<divclass="form-group">
        {{ form.body.label }}<br>
        {{ form.body(rows=15, cols=80) }}<br>
        {% for error in form.body.errors %}
<spanclass="error">{{ error }}</span>
        {% endfor %}
</div>
<divclass="form-group">
        {{ form.submit() }}
</div>
</form>
{% endblock %}

13. 静态文件 static/style.css（简单示例）