Flask 入门教程 —— 从零构建 Web 应用
本章目标
- 理解 Flask 的核心设计理念
- 搭建完整的 Flask 项目骨架
- 掌握路由、模板、静态文件三大基础
- 实现一个简单的留言板应用
1. Flask 的设计哲学
Flask 遵循微内核 + 扩展的设计模式:
Flask 核心(最小集合)
├── 路由系统(werkzeug.routing)
├── 请求/响应封装(werkzeug.wrappers)
├── 模板渲染(Jinja2)
├── 开发服务器
└── 测试客户端
按需扩展(pip install 自由搭配)
├── flask-sqlalchemy → 数据库
├── flask-login → 认证
├── flask-admin → 后台管理
├── flask-socketio → WebSocket
└── ...上百个扩展
对比 Django:Django 是"大而全"的框架,自带 ORM、后台、表单、认证;Flask 是一张白纸,你可以自由选择每个组件。
2. 项目骨架
my_flask_app/
├── app/ # 应用包
│ ├── __init__.py # 工厂函数 create_app()
│ ├── routes.py # 路由视图
│ ├── models.py # 数据库模型
│ ├── forms.py # 表单定义
│ ├── templates/ # Jinja2 模板
│ │ ├── base.html
│ │ ├── index.html
│ │ └── post.html
│ └── static/ # 静态文件
│ ├── css/
│ └── js/
├── migrations/ # 数据库迁移
├── config.py # 配置类
├── requirements.txt # 依赖
└── run.py # 启动入口
3. 应用程序工厂模式
# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
db = SQLAlchemy()
migrate = Migrate()
def create_app():
app = Flask(__name__)
app.config.from_object("config.Config")
# 初始化扩展
db.init_app(app)
migrate.init_app(app, db)
# 注册蓝图
from app.routes import main_bp
app.register_blueprint(main_bp)
return app
为什么用工厂模式?
- 支持多环境配置(开发/测试/生产)
- 便于单元测试(每次
create_app() 都是全新实例)
- 延迟初始化,避免循环导入
4. 蓝图(Blueprint)组织路由
# app/routes.py
from flask import Blueprint, render_template, request, redirect, url_for
from app.models import Message
from app import db
main_bp = Blueprint("main", __name__)
@main_bp.route("/")
def index():
messages = Message.query.order_by(Message.created_at.desc()).all()
return render_template("index.html", messages=messages)
@main_bp.route("/post", methods=["POST"])
def post():
content = request.form.get("content", "").strip()
if content:
msg = Message(content=content)
db.session.add(msg)
db.session.commit()
return redirect(url_for("main.index"))
蓝图的好处
- 将大型应用拆分为独立的"迷你应用"
- 每个蓝图可有自己的模板/静态文件目录
- 支持 URL 前缀(如
/admin、/api/v1)
5. Jinja2 模板实战
<!-- app/templates/base.html -->
<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{% block title %}留言板{% endblock %}</title>
<link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
</head>
<body>
<div class="container">
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
<div class="alert alert-{{ category }}">{{ message }}</div>
{% endfor %}
{% endif %}
{% endwith %}
{% block content %}{% endblock %}
</div>
</body>
</html>
<!-- app/templates/index.html -->
{% extends "base.html" %}
{% block title %}留言板 - 首页{% endblock %}
{% block content %}
<h1>📝 留言板</h1>
<form method="post" action="{{ url_for('main.post') }}">
<textarea name="content" rows="3" placeholder="说点什么吧..." required></textarea>
<button type="submit">发布</button>
</form>
<hr>
{% for msg in messages %}
<div class="message">
<p>{{ msg.content }}</p>
<small>{{ msg.created_at.strftime('%Y-%m-%d %H:%M') }}</small>
</div>
{% else %}
<p>还没有留言,来写第一条吧!</p>
{% endfor %}
{% endblock %}
Jinja2 关键语法
| 语法 |
说明 |
{% block name %} |
定义可被子模板替换的块 |
{% extends "base.html" %} |
继承基础模板 |
{% for ... %} / {% endfor %} |
循环 |
{% if ... %} / {% endif %} |
条件判断 |
{% with ... %} |
变量作用域 |
{{ variable }} |
输出变量(自动 HTML 转义) |
{{ | safe }} |
不转义输出(慎用) |
url_for() |
反向生成 URL(路由改名时自动更新) |
6. 使用 Flask-Login 实现用户认证
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
login_manager = LoginManager()
login_manager.login_view = "auth.login"
class User(db.Model, UserMixin):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), unique=True, nullable=False)
password_hash = db.Column(db.String(128), nullable=False)
def set_password(self, password):
from werkzeug.security import generate_password_hash
self.password_hash = generate_password_hash(password)
def check_password(self, password):
from werkzeug.security import check_password_hash
return check_password_hash(self.password_hash, password)
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
# 保护路由
@main_bp.route("/secret")
@login_required
def secret():
return f"只有登录用户能看到,你好 {current_user.username}!"
7. 配置管理最佳实践
# config.py
import os
from dotenv import load_dotenv
load_dotenv() # 加载 .env 文件
class Config:
SECRET_KEY = os.getenv("SECRET_KEY", "dev-key-change-me")
SQLALCHEMY_DATABASE_URI = os.getenv("DATABASE_URL", "sqlite:///app.db")
SQLALCHEMY_TRACK_MODIFICATIONS = False
JSON_AS_ASCII = False # 支持中文 JSON 输出
class DevelopmentConfig(Config):
DEBUG = True
class ProductionConfig(Config):
DEBUG = False
class TestingConfig(Config):
TESTING = True
SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"
思考题
- 蓝图与直接
@app.route 有何区别?何时应该使用蓝图?
url_for() 比硬编码 URL 有哪些优势?
- Flask 的"微"体现在哪里?为什么说这是优势而非劣势?
- 工厂模式
create_app() 相比全局 app = Flask(__name__) 有什么好处?
Flask 进阶实战 —— 异步任务、Docker 部署与性能优化
本章目标
- 掌握 Celery + Redis 异步任务队列
- 使用 Docker 容器化 Flask 应用
- 了解 Gunicorn 生产部署
- 常见性能优化策略
1. Celery 异步任务
1.1 为什么需要异步任务?
Web 请求应该在毫秒级完成。以下场景必须异步处理:
- 发送邮件验证码(可能耗时 2~5 秒)
- 生成报表/PDF(可能耗时 10 秒+)
- 图像/视频处理
- 调用第三方 API(超时风险)
1.2 架构概览
用户请求 → Flask (快速返回 202 Accepted)
│
↓
Redis/RabbitMQ (消息队列/代理)
│
↓
Celery Worker (后台异步执行)
│
↓
Redis/DB (结果存储)
│
↓
用户轮询/WebSocket 获取结果
1.3 配置与使用
# tasks.py
from celery import Celery
celery = Celery(
"my_flask_tasks",
broker="redis://localhost:6379/0", # 消息代理
backend="redis://localhost:6379/1", # 结果后端
)
# 配置序列化
celery.conf.update(
task_serializer="json",
accept_content=["json"],
result_expires=3600, # 结果过期时间(秒)
)
@celery.task(bind=True, max_retries=3)
def send_welcome_email(self, user_email, username):
"""发送欢迎邮件(带重试机制)"""
try:
# 模拟发送邮件
import smtplib
import time
time.sleep(2) # 模拟耗时操作
print(f"✅ 已发送欢迎邮件至 {user_email}")
return {"status": "ok", "email": user_email}
except Exception as exc:
print(f"❌ 发送失败,将重试: {exc}")
raise self.retry(exc=exc, countdown=60) # 60秒后重试
@celery.task
def generate_report(report_id):
"""生成报表——耗时任务"""
import time
time.sleep(5) # 模拟复杂计算
# ... 实际报表生成逻辑
return {"report_id": report_id, "status": "done"}
# app/routes.py 中使用异步任务
from tasks import send_welcome_email, generate_report
@main_bp.route("/register", methods=["POST"])
def register():
username = request.form["username"]
email = request.form["email"]
# 创建用户... (省略)
# 异步发送欢迎邮件
task = send_welcome_email.delay(email, username)
return jsonify({
"message": "注册成功,欢迎邮件正在发送中",
"task_id": task.id,
}), 202
@main_bp.route("/task/<task_id>")
def get_task_status(task_id):
"""查询异步任务状态"""
from celery.result import AsyncResult
result = AsyncResult(task_id, app=celery)
if result.state == "PENDING":
response = {"state": "PENDING", "progress": "等待执行..."}
elif result.state == "SUCCESS":
response = {"state": "SUCCESS", "result": result.get()}
elif result.state == "FAILURE":
response = {"state": "FAILURE", "error": str(result.info)}
else:
response = {"state": result.state}
return jsonify(response)
1.4 启动命令
sudo apt install redis-server
brew install redis
celery -A tasks.celery worker --loglevel=info --concurrency=4
celery -A tasks.celery beat --loglevel=info
celery -A tasks.celery worker -B --loglevel=info
2. Docker 容器化部署
FROM python:3.11-slim
WORKDIR /app
RUN apt-get update &
gcc \
&
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt -i https:
COPY . .
RUN useradd -m flaskuser &
USER flaskuser
EXPOSE 8000
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8000", "run:app"]
version: "3.8"
services:
web:
build: .
ports:
- "8000:8000"
environment:
- FLASK_ENV=production
- DATABASE_URL=postgresql:
- REDIS_URL=redis:
depends_on:
- db
- redis
db:
image: postgres:15-alpine
environment:
POSTGRES_USER: user
POSTGRES_PASSWORD: pass
POSTGRES_DB: flaskdb
volumes:
- postgres_data:/var/lib/postgresql/data
redis:
image: redis:7-alpine
celery:
build: .
command: celery -A tasks.celery worker --loglevel=info
environment:
- REDIS_URL=redis:
depends_on:
- redis
volumes:
postgres_data:
3. Gunicorn 生产部署
pip install gunicorn
gunicorn -w 4 -b 0.0.0.0:8000 "run:create_app()"
gunicorn \
-w 4 \
-b 0.0.0.0:8000 \
-k gevent \
--max-requests 10000 \
--max-requests-jitter 500 \
--access-logfile - \
--error-logfile - \
--log-level info \
"run:create_app()"
Worker 类型选择
| Worker 类型 |
适用场景 |
sync(默认) |
CPU 密集型、低并发 |
gevent |
IO 密集型、高并发(数据库查询、API 调用) |
gthread |
多线程,共享内存,适合中等并发 |
uvicorn.workers.UvicornWorker |
ASGI,FastAPI 部署(Flask 2.0+ 支持 async) |
4. 性能优化清单
4.1 数据库层面
# 使用连接池(Flask-SQLAlchemy 默认已具备)
app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {
"pool_size": 10, # 连接池大小
"pool_recycle": 3600, # 连接回收时间
"pool_pre_ping": True, # 使用前检查连接有效性
}
# N+1 查询优化 —— 使用 joinedload
# ❌ 坏:每条 Post 触发一次 User 查询
posts = Post.query.all()
# ✅ 好:一次 JOIN 查询全部加载
from sqlalchemy.orm import joinedload
posts = Post.query.options(joinedload(Post.author)).all()
4.2 缓存策略
# Flask-Caching
from flask_caching import Cache
cache = Cache(config={"CACHE_TYPE": "RedisCache", "CACHE_REDIS_URL": "redis://localhost:6379/2"})
@main_bp.route("/hot-posts")
@cache.cached(timeout=300) # 缓存 5 分钟
def hot_posts():
return jsonify(get_expensive_data())
4.3 静态文件优化
# Nginx 配置:让 Nginx 直接服务静态文件,不经过 Flask
location /static/ {
alias /app/static/;
expires 30d; # 设置缓存
add_header Cache-Control "public, immutable";
}
location / {
proxy_pass http:
}
思考题
- Celery 的
broker 和 backend 各起什么作用?可以用同一个 Redis 实例吗?
- 为什么生产环境不推荐用
flask run,而要用 Gunicorn?
- Docker Compose 中
depends_on 是否保证服务已就绪?如何实现真正的启动顺序控制?
- 什么场景适合用
gevent worker 而非默认 sync worker?