# Flask **Repository Path**: wangjllong/Flask ## Basic Information - **Project Name**: Flask - **Description**: Flask 从入门到精通 - **Primary Language**: Python - **License**: Apache-2.0 - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 2 - **Forks**: 1 - **Created**: 2025-11-13 - **Last Updated**: 2026-01-15 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # Python 3.12 Flask 从入门到精通 ## 目录 1. [环境准备](#1-环境准备) 2. [Flask 入门基础](#2-flask-入门基础) 3. [核心知识点详解](#3-核心知识点详解) 4. [进阶知识点详解](#4-进阶知识点详解) 5. [完整项目实战（博客系统）](#5-完整项目实战博客系统) 6. [项目部署与运行](#6-项目部署与运行) ## 1. 环境准备 ### 1.1 安装 Python 3.12 从 [Python 官网](https://www.python.org/downloads/release/python-3120/) 下载并安装 Python 3.12，配置环境变量（勾选 "Add Python to PATH"）。 验证安装： ```bash python --version # 或 python3.12 --version # 输出：Python 3.12.x ``` ### 1.2 创建虚拟环境 ```bash # 创建虚拟环境（Windows/Linux/Mac 通用） python -m venv flask-env # 激活虚拟环境 # Windows（cmd）： flask-env\Scripts\activate.bat # Windows（PowerShell）： .\flask-env\Scripts\Activate.ps1 # Linux/Mac： source flask-env/bin/activate ``` ### 1.3 安装依赖包 创建 `requirements.txt` 文件，写入依赖： ```txt Flask==2.3.3 pymysql==1.1.0 DBUtils==2.0.3 # 数据库连接池 redis==5.0.1 # Redis 客户端 PyJWT==2.8.0 # JWT 认证 Flask-WTF==1.2.1 # 表单验证 python-dotenv==1.0.0 # 环境变量管理 logging==0.5.1.2 # 日志（Python 内置） ``` 安装依赖： ```bash pip install -r requirements.txt ``` ## 2. Flask 入门基础 ### 2.1 第一个 Flask 应用（Hello World） 创建 `hello.py`： ```python from flask import Flask # 初始化 Flask 应用 app = Flask(__name__) # __name__ 表示当前模块名，Flask 用它定位静态文件和模板目录 # 路由装饰器：绑定 URL 路径与视图函数 @app.route('/') def hello_world(): return 'Hello Flask! Python 3.12 🚀' if __name__ == '__main__': # 运行应用（debug=True 开启调试模式） app.run(debug=True, host='0.0.0.0', port=5000) ``` ### 2.2 运行与访问 ```bash python hello.py ``` 访问 `http://localhost:5000`，页面显示 `Hello Flask! Python 3.12 🚀`。 ### 2.3 关键概念说明 - **应用实例（app）**：`Flask(__name__)` 创建的核心对象，管理所有配置、路由、扩展。 - **路由（@app.route）**：将 URL 路径映射到视图函数，支持动态路径、HTTP 方法限制。 - **视图函数**：处理请求并返回响应（字符串、HTML、JSON 等）。 - **调试模式（debug=True）**： - 代码修改后自动重启应用 - 显示错误堆栈信息 - 仅开发环境使用，生产环境禁用 ## 3. 核心知识点详解 ### 3.1 应用初始化与配置管理 Flask 配置支持多种加载方式，推荐使用环境变量 + 配置文件分离不同环境（开发/测试/生产）。 #### 3.1.1 配置加载方式 创建 `config.py`： ```python import os from dotenv import load_dotenv # 加载 .env 文件中的环境变量 load_dotenv() class Config: """基础配置类""" SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key') # 加密密钥（生产环境需修改） DEBUG = False TESTING = False class DevelopmentConfig(Config): """开发环境配置""" DEBUG = True # 数据库连接配置（后续用于连接池） MYSQL_HOST = os.getenv('MYSQL_HOST', 'localhost') MYSQL_PORT = int(os.getenv('MYSQL_PORT', 3306)) MYSQL_USER = os.getenv('MYSQL_USER', 'root') MYSQL_PASSWORD = os.getenv('MYSQL_PASSWORD', '123456') MYSQL_DB = os.getenv('MYSQL_DB', 'flask_blog') # Redis 配置 REDIS_HOST = os.getenv('REDIS_HOST', 'localhost') REDIS_PORT = int(os.getenv('REDIS_PORT', 6379)) REDIS_DB = int(os.getenv('REDIS_DB', 0)) class ProductionConfig(Config): """生产环境配置""" SECRET_KEY = os.getenv('SECRET_KEY') # 生产环境必须通过环境变量设置 MYSQL_HOST = os.getenv('MYSQL_HOST') MYSQL_PORT = int(os.getenv('MYSQL_PORT')) MYSQL_USER = os.getenv('MYSQL_USER') MYSQL_PASSWORD = os.getenv('MYSQL_PASSWORD') MYSQL_DB = os.getenv('MYSQL_DB') REDIS_HOST = os.getenv('REDIS_HOST') REDIS_PORT = int(os.getenv('REDIS_PORT')) REDIS_DB = int(os.getenv('REDIS_DB', 0)) # 配置映射，便于切换环境 config = { 'development': DevelopmentConfig, 'production': ProductionConfig, 'default': DevelopmentConfig } ``` 创建 `.env` 文件（不提交到版本控制）： ```env SECRET_KEY=your-dev-secret-key-123 MYSQL_HOST=localhost MYSQL_PORT=3306 MYSQL_USER=root MYSQL_PASSWORD=123456 MYSQL_DB=flask_blog REDIS_HOST=localhost REDIS_PORT=6379 ``` 初始化应用时加载配置： ```python from flask import Flask from config import config def create_app(config_name='default'): app = Flask(__name__) app.config.from_object(config[config_name]) # 加载配置类 return app # 运行应用 if __name__ == '__main__': app = create_app('development') app.run(host='0.0.0.0', port=5000) ``` ### 3.2 路由与视图 #### 3.2.1 基本路由 ```python @app.route('/') # 根路径 def index(): return '首页' @app.route('/about') # 静态路径 def about(): return '关于我们' ``` #### 3.2.2 动态路由（URL 参数） ```python # 字符串参数（默认） @app.route('/user/') def user_profile(username): return f'用户：{username}' # 整数参数 @app.route('/post/') def show_post(post_id): return f'文章 ID：{post_id}' # 仅接受整数 # 浮点数参数 @app.route('/price/') def show_price(amount): return f'价格：{amount}' # 路径参数（包含 /） @app.route('/file/') def show_file(file_path): return f'文件路径：{file_path}' ``` #### 3.2.3 HTTP 方法限制 默认路由仅支持 `GET` 方法，可通过 `methods` 参数指定： ```python from flask import request @app.route('/login', methods=['GET', 'POST']) def login(): if request.method == 'POST': # 处理表单提交 username = request.form.get('username') password = request.form.get('password') return f'登录提交：{username}' else: # 返回登录表单 return '''
用户名：
密码：
''' ``` #### 3.2.4 蓝图（Blueprint） 用于拆分大型应用，按功能模块组织路由（如用户模块、博客模块）。 创建 `app/blueprints/auth.py`（认证模块）： ```python from flask import Blueprint, render_template, request # 创建蓝图：参数（名称，模块名，模板文件夹） auth_bp = Blueprint('auth', __name__, template_folder='../templates/auth') # 蓝图路由（URL 前缀会自动添加 /auth） @auth_bp.route('/login', methods=['GET', 'POST']) def login(): if request.method == 'POST': # 处理登录逻辑 return '登录成功' return render_template('auth/login.html') @auth_bp.route('/register') def register(): return render_template('auth/register.html') ``` 创建 `app/blueprints/blog.py`（博客模块）： ```python from flask import Blueprint blog_bp = Blueprint('blog', __name__, template_folder='../templates/blog') @blog_bp.route('/') def index(): return '博客首页' @blog_bp.route('/post/') def show_post(post_id): return f'博客文章：{post_id}' ``` 在应用初始化时注册蓝图： ```python from app.blueprints.auth import auth_bp from app.blueprints.blog import blog_bp def create_app(config_name='default'): app = Flask(__name__) app.config.from_object(config[config_name]) # 注册蓝图（url_prefix 可选，添加 URL 前缀） app.register_blueprint(auth_bp, url_prefix='/auth') app.register_blueprint(blog_bp, url_prefix='/blog') return app ``` 访问路径： - 登录：`/auth/login` - 注册：`/auth/register` - 博客首页：`/blog/` - 博客文章：`/blog/post/1` ### 3.3 请求与响应 #### 3.3.1 请求对象（request） Flask 提供 `request` 对象封装 HTTP 请求信息： ```python from flask import request @app.route('/request-demo', methods=['GET', 'POST']) def request_demo(): # 1. 获取请求方法 method = request.method # 2. 获取 GET 参数（URL 中的 ?key=value） get_param = request.args.get('key', '默认值') # 3. 获取 POST 表单参数 form_param = request.form.get('username') # 4. 获取 JSON 数据（Content-Type: application/json） json_data = request.get_json() # 等同于 request.json # 5. 获取请求头 user_agent = request.headers.get('User-Agent') # 6. 获取 Cookie cookie = request.cookies.get('cookie_name') return { 'method': method, 'get_param': get_param, 'json_data': json_data, 'user_agent': user_agent } ``` #### 3.3.2 响应对象（response） ```python from flask import make_response, jsonify @app.route('/response-demo') def response_demo(): # 1. 基础响应（字符串） # return '响应内容', 200 # 状态码默认 200 # 2. JSON 响应（推荐用 jsonify） json_res = jsonify({'code': 200, 'msg': '成功', 'data': {'name': 'Flask'}}) json_res.status_code = 200 # 3. 自定义响应头 json_res.headers['X-Custom-Header'] = 'Flask-Demo' # 4. 设置 Cookie response = make_response(json_res) response.set_cookie( key='username', value='admin', max_age=3600, # 过期时间（秒） path='/', httponly=True # 禁止 JS 访问，防止 XSS ) return response ``` #### 3.3.3 重定向与 abort ```python from flask import redirect, url_for, abort @app.route('/redirect-demo') def redirect_demo(): # 重定向到指定 URL # return redirect('https://www.baidu.com') # 重定向到视图函数（推荐，自动处理 URL 前缀） return redirect(url_for('auth.login')) # auth 是蓝图名，login 是视图函数名 @app.route('/abort-demo/') def abort_demo(id): if id < 1: abort(404) # 抛出 404 错误 return f'ID：{id}' ``` ### 3.4 Jinja2 模板引擎 Flask 内置 Jinja2 模板引擎，用于渲染动态 HTML。 #### 3.4.1 模板目录结构 ``` app/ ├── templates/ │ ├── base.html # 基础模板（继承用） │ ├── auth/ │ │ ├── login.html │ │ └── register.html │ └── blog/ │ ├── index.html │ └── post.html ``` #### 3.4.2 模板继承（base.html） ```html {% block title %}Flask 博客{% endblock %}

Flask 博客系统

{% block content %}{% endblock %}

© 2025 Flask 博客系统

``` #### 3.4.2 子模板（login.html） ```html {% extends "base.html" %} {% block title %}登录 - Flask 博客{% endblock %} {% block content %} {% endblock %} ``` #### 3.4.3 模板语法 - **变量**：`{{ 变量名 }}`（支持 Python 表达式，如 `{{ user.name }}`、`{{ posts|length }}`） - **过滤器**：`{{ 变量|过滤器 }}`（如 `{{ text|upper }}`、`{{ date|datetimeformat('%Y-%m-%d') }}`） - **控制结构**： ```html {% for post in posts %}

{{ post.title }}

{{ post.content }}

{% else %}

暂无文章

{% endfor %} {% if user.role == 'admin' %} 发布文章 {% endif %} ``` - **宏（类似函数）**： ```html {% macro input(name, label, type='text') %}
{% endmacro %} {{ input('username', '用户名') }} {{ input('password', '密码', 'password') }} ``` ### 3.5 静态文件管理 静态文件（CSS、JS、图片）放在 `app/static` 目录，通过 `url_for('static', filename='路径')` 引用： ``` app/ ├── static/ │ ├── css/ │ │ └── style.css │ ├── js/ │ │ └── main.js │ └── images/ │ └── logo.png ``` 引用示例： ```html Logo ``` ## 4. 进阶知识点详解 ### 4.1 MySQL 数据库连接池 使用 `DBUtils.PooledDB` 实现 MySQL 连接池，避免频繁创建/销毁数据库连接，提升性能。 #### 4.1.1 初始化连接池 创建 `app/extensions.py`（管理扩展实例）： ```python from DBUtils.PooledDB import PooledDB import pymysql from redis import Redis from flask import current_app # 数据库连接池实例 db_pool = None # Redis 实例 redis_client = None def init_extensions(app): """初始化扩展（数据库连接池、Redis）""" global db_pool, redis_client # 初始化 MySQL 连接池 db_pool = PooledDB( creator=pymysql, # 数据库驱动 maxconnections=10, # 最大连接数 mincached=2, # 最小空闲连接数 maxcached=5, # 最大空闲连接数 maxshared=3, # 最大共享连接数（0 表示所有连接都是独占的） blocking=True, # 无空闲连接时是否阻塞等待 maxusage=None, # 单个连接最大使用次数（None 表示无限制） setsession=[], # 连接建立时执行的 SQL 语句 ping=0, # 检查连接可用性的频率（0 表示不检查） host=app.config['MYSQL_HOST'], port=app.config['MYSQL_PORT'], user=app.config['MYSQL_USER'], password=app.config['MYSQL_PASSWORD'], database=app.config['MYSQL_DB'], charset='utf8mb4' ) # 初始化 Redis 客户端 redis_client = Redis( host=app.config['REDIS_HOST'], port=app.config['REDIS_PORT'], db=app.config['REDIS_DB'], decode_responses=True # 自动解码为字符串（避免 b'xxx' 格式） ) def get_db_connection(): """获取数据库连接（从连接池）""" return db_pool.connection() ``` 在 `create_app` 中初始化扩展： ```python from app.extensions import init_extensions def create_app(config_name='default'): app = Flask(__name__) app.config.from_object(config[config_name]) # 初始化扩展（数据库连接池、Redis） init_extensions(app) # 注册蓝图... return app ``` #### 4.1.2 数据库操作示例 创建 `app/models/user.py`（用户数据模型）： ```python from app.extensions import get_db_connection class User: @staticmethod def get_by_username(username): """根据用户名查询用户""" conn = None try: conn = get_db_connection() with conn.cursor(pymysql.cursors.DictCursor) as cursor: sql = "SELECT id, username, password, email, role FROM users WHERE username = %s" cursor.execute(sql, (username,)) return cursor.fetchone() # 返回字典格式的用户数据 finally: if conn: conn.close() # 归还连接到连接池（不是关闭连接） @staticmethod def create(username, password, email): """创建新用户""" conn = None try: conn = get_db_connection() with conn.cursor() as cursor: sql = """ INSERT INTO users (username, password, email, role, created_at) VALUES (%s, %s, %s, 'user', NOW()) """ cursor.execute(sql, (username, password, email)) conn.commit() return cursor.lastrowid # 返回新用户 ID finally: if conn: conn.close() ``` ### 4.2 Redis 操作 Redis 用于存储 JWT Token、缓存等，已在 `extensions.py` 中初始化 `redis_client`。 #### 4.2.1 常用操作 ```python from app.extensions import redis_client # 1. 字符串操作（存储 JWT Token） redis_client.set('token:user:1', 'jwt-token-string', ex=3600) # ex=3600 过期时间（秒） token = redis_client.get('token:user:1') # 获取 Token redis_client.delete('token:user:1') # 删除 Token exists = redis_client.exists('token:user:1') # 检查是否存在 # 2. 哈希操作（存储用户信息缓存） redis_client.hset('user:1', mapping={'username': 'admin', 'role': 'admin'}) user_info = redis_client.hgetall('user:1') # 获取所有字段 # 3. 列表操作（存储最新文章 ID） redis_client.lpush('latest_posts', 1001, 1002, 1003) latest_posts = redis_client.lrange('latest_posts', 0, 9) # 获取前 10 条 ``` ### 4.3 JWT 认证（结合 Redis） JWT（JSON Web Token）用于无状态认证，结合 Redis 实现 Token 吊销（登出）功能。 #### 4.3.1 JWT 工具类 创建 `app/utils/jwt_utils.py`： ```python import jwt import time from flask import current_app from app.extensions import redis_client # Token 过期时间（3600 秒 = 1 小时） ACCESS_TOKEN_EXPIRE_SECONDS = 3600 def generate_token(user_id): """生成 JWT Token 并存入 Redis""" # 1. 构建 Token payload（负载） payload = { 'user_id': user_id, 'exp': int(time.time()) + ACCESS_TOKEN_EXPIRE_SECONDS, # 过期时间 'iat': int(time.time()) # 签发时间 } # 2. 生成 Token（使用 HS256 算法） token = jwt.encode( payload, current_app.config['SECRET_KEY'], algorithm='HS256' ) # 3. 存入 Redis（key: token:user:{user_id}, value: token, 过期时间与 Token 一致） redis_client.set( f'token:user:{user_id}', token, ex=ACCESS_TOKEN_EXPIRE_SECONDS ) return token def verify_token(token): """验证 Token 有效性（从 Redis 读取并验证签名）""" try: # 1. 解码 Token（验证签名和过期时间） payload = jwt.decode( token, current_app.config['SECRET_KEY'], algorithms=['HS256'] ) user_id = payload['user_id'] # 2. 检查 Redis 中是否存在该 Token（防止已登出） redis_token = redis_client.get(f'token:user:{user_id}') if not redis_token or redis_token != token: return None # Token 已吊销 return user_id # 验证通过，返回用户 ID except jwt.ExpiredSignatureError: # Token 过期 return None except jwt.InvalidTokenError: # Token 无效 return None def revoke_token(user_id): """吊销 Token（从 Redis 删除）""" redis_client.delete(f'token:user:{user_id}') ``` #### 4.3.2 认证装饰器 创建 `app/utils/auth_decorator.py`： ```python from functools import wraps from flask import request, jsonify, g from app.utils.jwt_utils import verify_token def login_required(f): """登录验证装饰器：必须携带有效 Token 才能访问""" @wraps(f) def decorated_function(*args, **kwargs): # 1. 从请求头获取 Token（格式：Authorization: Bearer ） auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'code': 401, 'msg': '未提供认证 Token'}), 401 token = auth_header.split(' ')[1] # 2. 验证 Token user_id = verify_token(token) if not user_id: return jsonify({'code': 401, 'msg': 'Token 无效或已过期'}), 401 # 3. 将用户 ID 存入 g 对象（全局可访问） g.user_id = user_id return f(*args, **kwargs) return decorated_function ``` #### 4.3.3 认证使用示例 ```python from flask import jsonify, g from app.utils.auth_decorator import login_required from app.models.user import User @auth_bp.route('/login', methods=['POST']) def login(): username = request.form.get('username') password = request.form.get('password') # 1. 查询用户（实际项目中密码需加密存储，这里简化为明文对比） user = User.get_by_username(username) if not user or user['password'] != password: return jsonify({'code': 400, 'msg': '用户名或密码错误'}), 400 # 2. 生成 Token token = generate_token(user['id']) return jsonify({ 'code': 200, 'msg': '登录成功', 'data': { 'user_id': user['id'], 'username': user['username'], 'token': token } }) @auth_bp.route('/logout', methods=['POST']) @login_required # 需登录才能访问 def logout(): # 吊销 Token revoke_token(g.user_id) return jsonify({'code': 200, 'msg': '登出成功'}) @blog_bp.route('/post/create', methods=['POST']) @login_required def create_post(): # g.user_id 为当前登录用户 ID user_id = g.user_id title = request.form.get('title') content = request.form.get('content') # 处理发布文章逻辑... return jsonify({'code': 200, 'msg': '文章发布成功'}) ``` ### 4.4 错误处理与日志 #### 4.4.1 全局错误处理 创建 `app/utils/error_handler.py`： ```python from flask import jsonify import logging # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('app.log'), # 写入文件 logging.StreamHandler() # 输出到控制台 ] ) logger = logging.getLogger(__name__) def register_error_handlers(app): """注册全局错误处理器""" # 404 错误 @app.errorhandler(404) def handle_404(e): logger.warning(f'404 错误：{e}') return jsonify({'code': 404, 'msg': '页面不存在'}), 404 # 500 错误 @app.errorhandler(500) def handle_500(e): logger.error(f'500 错误：{e}', exc_info=True) return jsonify({'code': 500, 'msg': '服务器内部错误'}), 500 # 自定义异常 @app.errorhandler(Exception) def handle_exception(e): logger.error(f'未捕获异常：{e}', exc_info=True) return jsonify({'code': 500, 'msg': '未知错误'}), 500 ``` 在 `create_app` 中注册错误处理器： ```python from app.utils.error_handler import register_error_handlers def create_app(config_name='default'): app = Flask(__name__) # ... 其他初始化 ... # 注册全局错误处理器 register_error_handlers(app) return app ``` #### 4.4.2 日志使用示例 ```python from app.utils.error_handler import logger @app.route('/demo-log') def demo_log(): logger.info('访问了日志演示接口') logger.warning('这是一个警告日志') try: 1 / 0 except Exception as e: logger.error('发生除零错误', exc_info=True) # exc_info=True 记录堆栈信息 return '日志演示' ``` ### 4.5 表单验证（Flask-WTF） 使用 `Flask-WTF` 处理表单验证，防止 CSRF 攻击。 #### 4.5.1 表单类 创建 `app/forms/auth.py`： ```python from flask_wtf import FlaskForm from wtforms import StringField, PasswordField, SubmitField from wtforms.validators import DataRequired, Length, Email, EqualTo class LoginForm(FlaskForm): """登录表单""" username = StringField('用户名', validators=[DataRequired('用户名不能为空'), Length(3, 20, '用户名长度需 3-20 字符')]) password = PasswordField('密码', validators=[DataRequired('密码不能为空'), Length(6, 20, '密码长度需 6-20 字符')]) submit = SubmitField('登录') class RegisterForm(FlaskForm): """注册表单""" username = StringField('用户名', validators=[DataRequired(), Length(3, 20)]) email = StringField('邮箱', validators=[DataRequired(), Email('邮箱格式不正确')]) password = PasswordField('密码', validators=[DataRequired(), Length(6, 20)]) confirm_password = PasswordField( '确认密码', validators=[DataRequired(), EqualTo('password', '两次密码不一致')] ) submit = SubmitField('注册') ``` #### 4.5.2 表单使用示例 ```python from app.forms.auth import LoginForm, RegisterForm from app.models.user import User from werkzeug.security import generate_password_hash # 密码加密 @auth_bp.route('/register', methods=['GET', 'POST']) def register(): form = RegisterForm() if form.validate_on_submit(): # 表单提交且验证通过 username = form.username.data email = form.email.data # 密码加密存储（绝不能存明文） password = generate_password_hash(form.password.data, method='pbkdf2:sha256') # 创建用户 user_id = User.create(username, password, email) return redirect(url_for('auth.login')) # 表单验证失败，获取错误信息 errors = {} for field, field_errors in form.errors.items(): errors[field] = field_errors[0] return render_template('auth/register.html', form=form, errors=errors) ``` > 注意：Flask-WTF 需配置 `SECRET_KEY` 用于 CSRF 令牌生成（已在 `config.py` 中配置）。 ### 4.6 文件上传 ```python import os from flask import request, jsonify, url_for from werkzeug.utils import secure_filename # 配置文件上传路径和允许的扩展名 UPLOAD_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), '../static/uploads') ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'} # 确保上传目录存在 os.makedirs(UPLOAD_FOLDER, exist_ok=True) def allowed_file(filename): """检查文件扩展名是否允许""" return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS @blog_bp.route('/upload', methods=['POST']) @login_required def upload_file(): # 检查是否有文件上传 if 'file' not in request.files: return jsonify({'code': 400, 'msg': '未选择文件'}), 400 file = request.files['file'] if file.filename == '': return jsonify({'code': 400, 'msg': '文件名不能为空'}), 400 if file and allowed_file(file.filename): # 安全处理文件名（防止路径遍历攻击） filename = secure_filename(file.filename) # 生成唯一文件名（避免覆盖） filename = f'{int(time.time())}_{filename}' # 保存文件 file_path = os.path.join(UPLOAD_FOLDER, filename) file.save(file_path) # 返回文件访问 URL file_url = url_for('static', filename=f'uploads/{filename}') return jsonify({'code': 200, 'msg': '上传成功', 'data': {'file_url': file_url}}) return jsonify({'code': 400, 'msg': '不允许的文件类型'}), 400 ``` ## 5. 完整项目实战（博客系统） ### 5.1 项目架构 ``` flask-blog/ ├── app/ │ ├── __init__.py # 应用工厂函数 │ ├── config.py # 配置文件 │ ├── extensions.py # 扩展初始化（数据库连接池、Redis） │ ├── blueprints/ # 蓝图模块 │ │ ├── __init__.py │ │ ├── auth.py # 认证模块（登录、注册、登出） │ │ └── blog.py # 博客模块（文章增删改查） │ ├── models/ # 数据模型（数据库操作） │ │ ├── __init__.py │ │ ├── user.py # 用户模型 │ │ └── post.py # 文章模型 │ ├── forms/ # 表单类 │ │ ├── __init__.py │ │ └── auth.py # 认证表单 │ ├── utils/ # 工具函数 │ │ ├── __init__.py │ │ ├── jwt_utils.py # JWT 工具 │ │ ├── auth_decorator.py # 认证装饰器 │ │ └── error_handler.py # 错误处理 │ ├── templates/ # 模板文件 │ │ ├── base.html # 基础模板 │ │ ├── auth/ │ │ │ ├── login.html │ │ │ └── register.html │ │ └── blog/ │ │ ├── index.html │ │ ├── post.html │ │ └── create_post.html │ └── static/ # 静态文件 │ ├── css/ │ │ └── style.css │ ├── js/ │ │ └── main.js │ └── uploads/ # 上传文件存储 ├── .env # 环境变量（不提交版本控制） ├── requirements.txt # 依赖包 ├── run.py # 应用入口 └── sql/ # 数据库脚本 └── init_db.sql # 初始化数据库表 ``` ### 5.2 核心代码实现 #### 5.2.1 应用入口（run.py） ```python from app import create_app app = create_app('development') if __name__ == '__main__': app.run(host='0.0.0.0', port=5000) ``` #### 5.2.2 应用工厂（app/__init__.py） ```python from flask import Flask from app.config import config from app.extensions import init_extensions from app.utils.error_handler import register_error_handlers from app.blueprints.auth import auth_bp from app.blueprints.blog import blog_bp def create_app(config_name='default'): app = Flask(__name__) app.config.from_object(config[config_name]) # 初始化扩展 init_extensions(app) # 注册错误处理器 register_error_handlers(app) # 注册蓝图 app.register_blueprint(auth_bp, url_prefix='/auth') app.register_blueprint(blog_bp, url_prefix='/blog') # 根路径重定向到博客首页 @app.route('/') def index(): from flask import redirect, url_for return redirect(url_for('blog.index')) return app ``` #### 5.2.3 数据库初始化脚本（sql/init_db.sql） ```sql CREATE DATABASE IF NOT EXISTS flask_blog CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE flask_blog; -- 用户表 CREATE TABLE IF NOT EXISTS users ( id INT PRIMARY KEY AUTO_INCREMENT, username VARCHAR(20) NOT NULL UNIQUE COMMENT '用户名', password VARCHAR(255) NOT NULL COMMENT '加密密码', email VARCHAR(100) NOT NULL UNIQUE COMMENT '邮箱', role ENUM('user', 'admin') DEFAULT 'user' COMMENT '角色', created_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间' ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 文章表 CREATE TABLE IF NOT EXISTS posts ( id INT PRIMARY KEY AUTO_INCREMENT, title VARCHAR(100) NOT NULL COMMENT '文章标题', content TEXT NOT NULL COMMENT '文章内容', author_id INT NOT NULL COMMENT '作者ID', cover_url VARCHAR(255) DEFAULT '' COMMENT '封面图URL', created_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', FOREIGN KEY (author_id) REFERENCES users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 插入测试数据（密码：123456，加密方式：pbkdf2:sha256） INSERT INTO users (username, password, email, role) VALUES ('admin', 'pbkdf2:sha256:600000$QeUJZ0a7$e1b7d4e4a9f5d3c2b1a09f8e7d6c5b4a39876543210f9e8d7c6b5a4987654321', 'admin@example.com', 'admin') ON DUPLICATE KEY UPDATE username=username; ``` #### 5.2.4 文章模型（app/models/post.py） ```python from app.extensions import get_db_connection class Post: @staticmethod def create(title, content, author_id, cover_url=''): """创建文章""" conn = None try: conn = get_db_connection() with conn.cursor() as cursor: sql = """ INSERT INTO posts (title, content, author_id, cover_url) VALUES (%s, %s, %s, %s) """ cursor.execute(sql, (title, content, author_id, cover_url)) conn.commit() return cursor.lastrowid finally: if conn: conn.close() @staticmethod def get_all(page=1, per_page=10): """获取所有文章（分页）""" offset = (page - 1) * per_page conn = None try: conn = get_db_connection() with conn.cursor(pymysql.cursors.DictCursor) as cursor: # 关联查询作者信息 sql = """ SELECT p.*, u.username as author_name FROM posts p JOIN users u ON p.author_id = u.id ORDER BY p.created_at DESC LIMIT %s OFFSET %s """ cursor.execute(sql, (per_page, offset)) posts = cursor.fetchall() # 查询总数量 cursor.execute("SELECT COUNT(*) as total FROM posts") total = cursor.fetchone()['total'] return { 'posts': posts, 'total': total, 'pages': (total + per_page - 1) // per_page, 'current_page': page } finally: if conn: conn.close() @staticmethod def get_by_id(post_id): """根据ID获取文章""" conn = None try: conn = get_db_connection() with conn.cursor(pymysql.cursors.DictCursor) as cursor: sql = """ SELECT p.*, u.username as author_name FROM posts p JOIN users u ON p.author_id = u.id WHERE p.id = %s """ cursor.execute(sql, (post_id,)) return cursor.fetchone() finally: if conn: conn.close() @staticmethod def update(post_id, title, content, cover_url=''): """更新文章""" conn = None try: conn = get_db_connection() with conn.cursor() as cursor: sql = """ UPDATE posts SET title = %s, content = %s, cover_url = %s, updated_at = NOW() WHERE id = %s """ cursor.execute(sql, (title, content, cover_url, post_id)) conn.commit() return cursor.rowcount > 0 # 返回是否更新成功 finally: if conn: conn.close() @staticmethod def delete(post_id, author_id): """删除文章（验证作者）""" conn = None try: conn = get_db_connection() with conn.cursor() as cursor: sql = "DELETE FROM posts WHERE id = %s AND author_id = %s" cursor.execute(sql, (post_id, author_id)) conn.commit() return cursor.rowcount > 0 finally: if conn: conn.close() ``` #### 5.2.5 博客蓝图（app/blueprints/blog.py） ```python from flask import Blueprint, render_template, request, jsonify, g, redirect, url_for from app.utils.auth_decorator import login_required from app.models.post import Post from app.models.user import User import os import time from werkzeug.utils import secure_filename blog_bp = Blueprint('blog', __name__, template_folder='../templates/blog') # 文件上传配置 UPLOAD_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), '../../static/uploads') ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'} os.makedirs(UPLOAD_FOLDER, exist_ok=True) def allowed_file(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS # 博客首页（分页展示文章） @blog_bp.route('/') def index(): page = request.args.get('page', 1, type=int) per_page = 5 result = Post.get_all(page, per_page) return render_template('blog/index.html', **result) # 文章详情页 @blog_bp.route('/post/') def show_post(post_id): post = Post.get_by_id(post_id) if not post: return render_template('404.html'), 404 return render_template('blog/post.html', post=post) # 创建文章页面（需登录） @blog_bp.route('/post/create', methods=['GET']) @login_required def create_post_page(): return render_template('blog/create_post.html') # 提交文章（API） @blog_bp.route('/post', methods=['POST']) @login_required def create_post(): title = request.form.get('title') content = request.form.get('content') author_id = g.user_id # 处理封面图上传 cover_url = '' if 'cover' in request.files: file = request.files['cover'] if file and allowed_file(file.filename): filename = secure_filename(file.filename) filename = f'{int(time.time())}_{filename}' file_path = os.path.join(UPLOAD_FOLDER, filename) file.save(file_path) cover_url = url_for('static', filename=f'uploads/{filename}') # 创建文章 post_id = Post.create(title, content, author_id, cover_url) return jsonify({ 'code': 200, 'msg': '文章发布成功', 'data': {'post_id': post_id} }) # 编辑文章页面 @blog_bp.route('/post//edit', methods=['GET']) @login_required def edit_post_page(post_id): post = Post.get_by_id(post_id) if not post or post['author_id'] != g.user_id: return render_template('404.html'), 404 return render_template('blog/edit_post.html', post=post) # 更新文章（API） @blog_bp.route('/post/', methods=['PUT']) @login_required def update_post(post_id): title = request.form.get('title') content = request.form.get('content') author_id = g.user_id # 处理封面图上传 cover_url = request.form.get('old_cover') # 保留旧封面 if 'cover' in request.files: file = request.files['cover'] if file and allowed_file(file.filename): filename = secure_filename(file.filename) filename = f'{int(time.time())}_{filename}' file_path = os.path.join(UPLOAD_FOLDER, filename) file.save(file_path) cover_url = url_for('static', filename=f'uploads/{filename}') # 更新文章 success = Post.update(post_id, title, content, cover_url) if success: return jsonify({'code': 200, 'msg': '文章更新成功'}) return jsonify({'code': 400, 'msg': '文章更新失败'}), 400 # 删除文章（API） @blog_bp.route('/post/', methods=['DELETE']) @login_required def delete_post(post_id): success = Post.delete(post_id, g.user_id) if success: return jsonify({'code': 200, 'msg': '文章删除成功'}) return jsonify({'code': 400, 'msg': '文章删除失败'}), 400 ``` #### 5.2.6 认证蓝图（app/blueprints/auth.py） ```python from flask import Blueprint, render_template, request, jsonify, redirect, url_for, g from app.forms.auth import LoginForm, RegisterForm from app.models.user import User from app.utils.jwt_utils import generate_token, revoke_token from app.utils.auth_decorator import login_required from werkzeug.security import generate_password_hash, check_password_hash auth_bp = Blueprint('auth', __name__, template_folder='../templates/auth') # 登录页面 @auth_bp.route('/login', methods=['GET']) def login_page(): return render_template('auth/login.html') # 登录接口 @auth_bp.route('/login', methods=['POST']) def login(): form = LoginForm(formdata=request.form) if not form.validate(): errors = {field: errors[0] for field, errors in form.errors.items()} return jsonify({'code': 400, 'msg': '表单验证失败', 'errors': errors}), 400 username = form.username.data password = form.password.data # 查询用户 user = User.get_by_username(username) if not user or not check_password_hash(user['password'], password): return jsonify({'code': 400, 'msg': '用户名或密码错误'}), 400 # 生成 Token token = generate_token(user['id']) return jsonify({ 'code': 200, 'msg': '登录成功', 'data': { 'user_id': user['id'], 'username': user['username'], 'role': user['role'], 'token': token } }) # 注册页面 @auth_bp.route('/register', methods=['GET']) def register_page(): return render_template('auth/register.html') # 注册接口 @auth_bp.route('/register', methods=['POST']) def register(): form = RegisterForm(formdata=request.form) if not form.validate(): errors = {field: errors[0] for field, errors in form.errors.items()} return jsonify({'code': 400, 'msg': '表单验证失败', 'errors': errors}), 400 username = form.username.data email = form.email.data password = generate_password_hash(form.password.data, method='pbkdf2:sha256') # 检查用户名是否已存在 if User.get_by_username(username): return jsonify({'code': 400, 'msg': '用户名已存在'}), 400 # 创建用户 user_id = User.create(username, password, email) return jsonify({ 'code': 200, 'msg': '注册成功', 'data': {'user_id': user_id} }) # 登出接口 @auth_bp.route('/logout', methods=['POST']) @login_required def logout(): revoke_token(g.user_id) return jsonify({'code': 200, 'msg': '登出成功'}) # 个人中心 @auth_bp.route('/profile') @login_required def profile(): user = User.get_by_id(g.user_id) return render_template('auth/profile.html', user=user) ``` ### 5.3 模板文件示例（关键部分） #### 5.3.1 博客首页（templates/blog/index.html） ```html {% extends "base.html" %} {% block title %}博客首页 - Flask 博客{% endblock %} {% block content %}

最新文章

{% if g.user_id %} 发布文章 {% endif %}
{% for post in posts %}
{% if post.cover_url %} {{ post.title }} {% endif %}

{{ post.title }}

作者：{{ post.author_name }} | 发布时间：{{ post.created_at.strftime('%Y-%m-%d %H:%M') }}

{{ post.content[:100] }}...

阅读全文
{% else %}
暂无文章，快去发布第一篇吧！
{% endfor %}
{% endblock %} ``` #### 5.3.2 发布文章页面（templates/blog/create_post.html） ```html {% extends "base.html" %} {% block title %}发布文章 - Flask 博客{% endblock %} {% block content %}

发布新文章

取消
{% endblock %} ``` ## 6. 项目部署与运行 ### 6.1 环境准备 1. 安装 MySQL 5.7+/8.0 和 Redis（本地或远程均可） 2. 执行数据库初始化脚本： ```bash # 登录 MySQL mysql -u root -p # 执行脚本 source sql/init_db.sql ``` 3. 确保 Redis 服务已启动： ```bash # Linux/Mac redis-server # Windows（需安装 Redis） redis-server.exe ``` ### 6.2 运行项目 ```bash # 激活虚拟环境 source flask-env/bin/activate # Linux/Mac # 或 flask-env\Scripts\activate.bat # Windows # 运行应用 python run.py ``` ### 6.3 功能测试 1. 访问 `http://localhost:5000` 进入博客首页 2. 注册用户：`http://localhost:5000/auth/register` 3. 登录：`http://localhost:5000/auth/login`（登录成功后返回 Token） 4. 发布文章：登录后访问 `http://localhost:5000/blog/post/create` 5. 测试文章编辑、删除、分页等功能 ### 6.4 生产环境部署（简要说明） 1. 禁用调试模式：`config_name='production'` 2. 使用 Gunicorn 作为 WSGI 服务器： ```bash pip install gunicorn gunicorn -w 4 -b 0.0.0.0:5000 "run:app" ``` 3. 配置 Nginx 反向代理（处理静态文件、负载均衡） 4. 设置环境变量（生产环境密钥、数据库配置等，避免硬编码） ## 总结 本教程从 Flask 基础入门，逐步深入核心知识点（路由、模板、请求响应），再到进阶特性（数据库连接池、Redis、JWT 认证、表单验证、文件上传），最后通过完整的博客系统实战，整合所有知识点。项目架构模块化、代码遵循 PEP8 规范，支持用户认证、文章管理、文件上传等核心功能，可直接运行并扩展。 后续可扩展方向： - 增加评论功能 - 实现文章分类和标签 - 添加全文搜索（Elasticsearch） - 优化前端界面（使用 Vue/React） - 增加权限精细化控制