十一:用 Flask 写后端——Python 版的 Express

一、回顾与本篇目标

在《后端零基础入门》系列中,你用 Node.js 和 Express 框架写出了完整的后端服务——路由、中间件、数据库操作、用户认证。你已经理解了后端开发的核心概念:接收 HTTP 请求、处理业务逻辑、查询数据库、返回 JSON 响应。

现在把这些概念迁移到 Python 世界。Python 也有自己的后端框架,最主流的有两个:Flask(轻量级,适合学习和中小项目)和 Django(功能全面,适合大型项目)。这一篇我们学 Flask——它和 Express 的设计理念非常接近:轻量、灵活、上手快,一个文件就能跑起来。

如果你之前学过 Express,这篇的学习速度会非常快——核心概念完全一样,只是语法从 JavaScript 换成了 Python。

本篇的目标:

  1. 安装 Flask 并启动第一个 Flask 服务
  2. 理解 Flask 的路由定义方式
  3. 学会获取请求参数:路径参数、查询参数、请求体
  4. 学会返回 JSON 响应
  5. 把之前的留言板用户系统用 Flask 重写一遍

二、Flask 是什么

Flask 是一个用 Python 写的轻量级 Web 框架。它由一个核心库加上丰富的扩展生态组成。和 Express 一样,Flask 只提供最基础的 HTTP 处理能力——路由、请求、响应——其他的功能(数据库、认证、表单验证)通过扩展来实现。

Flask 和 Express 的对应关系:

概念 Express (Node.js) Flask (Python)
创建应用 const app = express() app = Flask(__name__)
定义路由 app.get('/path', handler) @app.route('/path', methods=['GET'])
路径参数 /user/:idreq.params.id /user/<int:id> → 直接作为函数参数
查询参数 req.query.keyword request.args.get('keyword')
JSON 请求体 req.body(需 express.json() request.get_json()
返回 JSON res.json(data) return jsonify(data)
启动服务 app.listen(3000, callback) app.run(port=5000)

三、安装 Flask

pip install flask

验证安装:

python -c "import flask; print(flask.__version__)"

四、第一个 Flask 服务

和学 Express 时一样,从最经典的“Hello World”开始。新建 app.py

# app.py —— 第一个 Flask 服务

from flask import Flask

# 创建 Flask 应用
app = Flask(__name__)

# 定义路由:访问根路径时返回一段文字
@app.route('/')
def home():
    return '<h1>你好,这是 Flask 返回的内容!</h1>'

# 启动服务
if __name__ == '__main__':
    app.run(port=5000, debug=True)

在终端执行 python app.py,你会看到:

 * Running on http://127.0.0.1:5000

打开浏览器访问 http://localhost:5000,看到标题文字,说明 Flask 服务启动成功。

逐行解释:

  • Flask(__name__):创建一个 Flask 应用实例。__name__ 是当前模块名,Flask 用它来定位静态文件和模板的位置。这相当于 Express 的 express()
  • @app.route('/'):这是 Python 的装饰器语法。它的意思是:把下面这个函数绑定到 / 路径上。当用户访问 / 时,Flask 会调用这个函数,把返回值作为 HTTP 响应发给浏览器。这和 Express 的 app.get('/', handler) 是同一个概念,只是写法不同。
  • return '<h1>...</h1>':Flask 自动把字符串包装成 HTTP 响应,默认 Content-Type: text/html
  • app.run(port=5000, debug=True):启动开发服务器。port=5000 指定端口(Flask 默认 5000,Express 默认 3000)。debug=True 开启调试模式,修改代码后服务器会自动重启,出错时浏览器会显示详细的错误信息。这相当于 Node.js 的 nodemon注意:生产环境不要开 debug=True。

关于装饰器

如果你之前没接触过 Python 的装饰器,可能会觉得 @app.route('/') 这种写法有点奇怪。装饰器本质上就是一个函数,它接收另一个函数作为参数,并返回一个新的函数

@app.route('/')
def home():
    return 'Hello'

# 上面的代码等价于:
# def home():
#     return 'Hello'
# home = app.route('/')(home)

你不需要完全理解装饰器的底层原理,只需要记住:在函数上面写 @app.route('/路径'),这个函数就变成了一个路由处理函数。

五、定义多种路由

Flask 用 methods 参数来限制路由接受的 HTTP 方法。默认只接受 GET 请求:

from flask import Flask

app = Flask(__name__)

# GET 请求(默认)
@app.route('/')
def home():
    return '<h1>首页</h1>'

# 同时接受 GET 和 POST
@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'POST':
        return '处理注册逻辑'
    return '<form method="POST"><button type="submit">注册</button></form>'

# 只接受 POST
@app.route('/api/data', methods=['POST'])
def handle_data():
    return '收到 POST 请求'

# 返回 JSON 数据
@app.route('/api/users')
def get_users():
    users = [
        {'id': 1, 'name': '张三'},
        {'id': 2, 'name': '李四'},
        {'id': 3, 'name': '王五'}
    ]
    return {'users': users}  # Flask 自动把字典转成 JSON

if __name__ == '__main__':
    app.run(port=5000, debug=True)

关键点:

  • methods=['GET', 'POST']:不写这个参数时,路由只接受 GET 请求。POST 请求会返回 405 Method Not Allowed。
  • 返回字典:Flask 会自动把 return {'key': 'value'} 这样的字典转换成 JSON 字符串,并设置 Content-Type: application/json在旧版 Flask 中需要用 jsonify(),新版可以直接返回字典。
  • 处理不同方法的逻辑:在函数内部用 request.method 来判断当前是 GET 还是 POST,这和 Express 中分别写 app.get()app.post() 的思路不同——Flask 把同一个路径的不同方法放在同一个函数里处理。

六、获取请求参数

后端开发的核心就是处理请求中的参数。Flask 提供了简洁的方式来获取各种类型的参数。

6.1 路径参数

在路由中用 <类型:变量名> 来定义路径参数,Flask 会自动把它作为函数参数传入:

# 路径参数:/user/123
@app.route('/user/<int:user_id>')
def get_user(user_id):
    return f'你请求的用户 ID 是:{user_id}(类型:{type(user_id).__name__})'

# 字符串类型的路径参数
@app.route('/article/<string:slug>')
def get_article(slug):
    return f'你请求的文章标识是:{slug}'

# 不指定类型时,默认是字符串
@app.route('/tag/<tag_name>')
def get_tag(tag_name):
    return f'标签:{tag_name}'

和 Express 的对比:

// Express: /user/:id → req.params.id(始终是字符串)
// Flask:   /user/<int:id> → id 已经是整数类型,不需要手动转换

Flask 支持的类型转换器:string(默认)、intfloatpath(包含斜杠的路径)、uuid

6.2 查询参数

查询参数就是 URL 中 ? 后面的键值对,通过 request.args.get() 获取:

from flask import Flask, request

@app.route('/search')
def search():
    # request.args 是一个字典,包含所有查询参数
    keyword = request.args.get('keyword', '')      # 第二个参数是默认值
    page = request.args.get('page', 1, type=int)    # type=int 自动转成整数
    return f'搜索关键词:{keyword},页码:{page}'

访问 http://localhost:5000/search?keyword=Python&page=2,输出:搜索关键词:Python,页码:2

和 Express 的对比:

// Express: req.query.keyword
# Flask:   request.args.get('keyword')

request.args.get() 比直接用 request.args['keyword'] 更安全——如果参数不存在,get() 返回 None(或你指定的默认值),而方括号写法会抛出异常。

6.3 JSON 请求体

当客户端(浏览器或 Postman)发送 JSON 格式的请求体时,用 request.get_json() 解析:

from flask import Flask, request

@app.route('/api/user', methods=['POST'])
def create_user():
    # 解析 JSON 请求体
    data = request.get_json()

    if not data:
        return {'error': '请求体不是有效的 JSON'}, 400

    name = data.get('name')
    email = data.get('email')

    if not name or not email:
        return {'error': '缺少必填字段:name 和 email'}, 400

    # 处理逻辑...
    return {
        'message': '用户创建成功',
        'user': {'name': name, 'email': email}
    }, 201

关键点:

  • request.get_json() 解析 JSON 请求体,返回 Python 字典。和 Express 的 express.json() 中间件类似,但 Flask 不需要显式注册中间件——它自动处理 JSON 请求。
  • 返回元组 (data, status_code):Flask 允许在 return 后面跟一个状态码,作为元组的第二个元素。return data, 201 等价于 Express 的 res.status(201).json(data)

6.4 表单数据

如果客户端提交的是 HTML 表单(Content-Type: application/x-www-form-urlencoded),用 request.form

@app.route('/login', methods=['POST'])
def login():
    username = request.form.get('username')
    password = request.form.get('password')
    return f'收到登录请求:{username}'

七、返回 JSON 响应

返回 JSON 是后端 API 最常见的操作。Flask 有三种方式:

# 方式一:直接返回字典(Flask 2.2+ 支持,最简洁)
@app.route('/api/user/1')
def get_user():
    return {'id': 1, 'name': '张三', 'email': 'zhangsan@example.com'}

# 方式二:用 jsonify()(旧版 Flask 兼容,仍广泛使用)
from flask import jsonify

@app.route('/api/user/2')
def get_user2():
    return jsonify(id=2, name='李四', email='lisi@example.com')

# 方式三:返回带状态码的响应
@app.route('/api/error')
def error_demo():
    return {'error': '资源不存在'}, 404

八、CORS 跨域处理

前后端分离开发时,前端页面(如 localhost:3000)调用后端 API(localhost:5000)属于跨域请求,浏览器会阻止。需要在后端设置 CORS 响应头:

pip install flask-cors
from flask import Flask
from flask_cors import CORS

app = Flask(__name__)
CORS(app)  # 允许所有来源的跨域请求

# 或者只允许特定来源
# CORS(app, origins=['http://localhost:3000'])

有了这行代码,前端就可以无障碍地调用这个 Flask 后端了。

九、实战:用 Flask 重写留言板用户系统

下面我们用 Flask 把《后端零基础入门》中的用户注册登录功能重写一遍。如果你之前写过 Node.js 版本,对比着看会发现思路完全一样。

项目结构

flask_message_board/
├── app.py              ← 主入口
├── requirements.txt    ← 依赖列表
└── .env                ← 环境变量

安装依赖

pip install flask flask-cors pymysql bcrypt pyjwt python-dotenv

依赖说明:

  • flask:Web 框架
  • flask-cors:跨域处理
  • pymysql:连接 MySQL
  • bcrypt:密码哈希
  • pyjwt:JWT 生成和验证
  • python-dotenv:加载 .env 环境变量

requirements.txt

flask==3.1.0
flask-cors==5.0.1
pymysql==1.1.1
bcrypt==4.2.1
pyjwt==2.10.1
python-dotenv==1.1.0

.env 文件

DB_HOST=localhost
DB_USER=root
DB_PASSWORD=你的数据库密码
DB_NAME=message_board
JWT_SECRET=change-this-to-a-random-string

完整代码:app.py

import os
import pymysql
import bcrypt
import jwt
from datetime import datetime, timedelta
from flask import Flask, request
from flask_cors import CORS
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

app = Flask(__name__)
CORS(app)

# ========== 数据库连接 ==========
def get_db_connection():
    """获取数据库连接"""
    return pymysql.connect(
        host=os.getenv('DB_HOST'),
        user=os.getenv('DB_USER'),
        password=os.getenv('DB_PASSWORD'),
        database=os.getenv('DB_NAME'),
        charset='utf8mb4',
        cursorclass=pymysql.cursors.DictCursor  # 返回字典格式的结果
    )

# ========== 用户注册 ==========
@app.route('/api/register', methods=['POST'])
def register():
    data = request.get_json()
    name = data.get('name')
    email = data.get('email')
    password = data.get('password')

    # 验证
    if not name or not email or not password:
        return {'error': '缺少必填字段:name、email、password'}, 400

    # 哈希加密密码
    hashed = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())

    try:
        conn = get_db_connection()
        with conn.cursor() as cursor:
            # 检查邮箱是否已注册
            cursor.execute('SELECT id FROM users WHERE email = %s', (email,))
            if cursor.fetchone():
                return {'error': '该邮箱已被注册'}, 409

            # 插入新用户
            cursor.execute(
                'INSERT INTO users (name, email, password) VALUES (%s, %s, %s)',
                (name, email, hashed.decode('utf-8'))
            )
            conn.commit()
            user_id = cursor.lastrowid

        # 生成 Token
        token = jwt.encode(
            {'user_id': user_id, 'email': email, 'exp': datetime.utcnow() + timedelta(days=7)},
            os.getenv('JWT_SECRET'),
            algorithm='HS256'
        )

        return {
            'message': '注册成功',
            'token': token,
            'user': {'id': user_id, 'name': name, 'email': email}
        }, 201

    except pymysql.Error as e:
        return {'error': f'数据库错误:{str(e)}'}, 500
    finally:
        conn.close()

# ========== 用户登录 ==========
@app.route('/api/login', methods=['POST'])
def login():
    data = request.get_json()
    email = data.get('email')
    password = data.get('password')

    if not email or not password:
        return {'error': '缺少邮箱或密码'}, 400

    try:
        conn = get_db_connection()
        with conn.cursor() as cursor:
            cursor.execute('SELECT * FROM users WHERE email = %s', (email,))
            user = cursor.fetchone()

        if not user:
            return {'error': '邮箱或密码错误'}, 401

        # 验证密码
        if not bcrypt.checkpw(password.encode('utf-8'), user['password'].encode('utf-8')):
            return {'error': '邮箱或密码错误'}, 401

        # 生成 Token
        token = jwt.encode(
            {'user_id': user['id'], 'email': user['email'], 'exp': datetime.utcnow() + timedelta(days=7)},
            os.getenv('JWT_SECRET'),
            algorithm='HS256'
        )

        return {
            'message': '登录成功',
            'token': token,
            'user': {'id': user['id'], 'name': user['name'], 'email': user['email']}
        }

    except pymysql.Error as e:
        return {'error': f'数据库错误:{str(e)}'}, 500
    finally:
        conn.close()

# ========== 认证中间件(作为装饰器) ==========
def login_required(func):
    """装饰器:验证 JWT Token"""
    def wrapper(*args, **kwargs):
        auth_header = request.headers.get('Authorization', '')
        if not auth_header.startswith('Bearer '):
            return {'error': '未登录,请先登录'}, 401

        token = auth_header.split(' ')[1]
        try:
            payload = jwt.decode(token, os.getenv('JWT_SECRET'), algorithms=['HS256'])
            request.user = payload  # 把解析出来的用户信息挂到 request 上
        except jwt.ExpiredSignatureError:
            return {'error': 'Token 已过期,请重新登录'}, 401
        except jwt.InvalidTokenError:
            return {'error': 'Token 无效'}, 401

        return func(*args, **kwargs)
    wrapper.__name__ = func.__name__  # 保留原函数名
    return wrapper

# ========== 获取当前用户信息(需登录) ==========
@app.route('/api/me')
@login_required
def get_me():
    try:
        conn = get_db_connection()
        with conn.cursor() as cursor:
            cursor.execute('SELECT id, name, email, created_at FROM users WHERE id = %s', (request.user['user_id'],))
            user = cursor.fetchone()

        if not user:
            return {'error': '用户不存在'}, 404

        # 格式化时间
        if user.get('created_at'):
            user['created_at'] = user['created_at'].strftime('%Y-%m-%d %H:%M:%S')

        return user
    except pymysql.Error as e:
        return {'error': f'数据库错误:{str(e)}'}, 500
    finally:
        conn.close()

# ========== 启动 ==========
if __name__ == '__main__':
    app.run(port=5000, debug=True)

代码中的关键差异(和 Node.js 版本对比):

  • 数据库连接:Python 没有连接池的标配(mysql2 自带),这里每次请求创建新连接。生产环境可以用 DBUtilsSQLAlchemy 来管理连接池。
  • cursorclass=DictCursor:让查询结果以字典形式返回(默认是元组),访问列时用 row['name'] 而不是 row[0]
  • with conn.cursor():Python 的上下文管理器,自动关闭游标。
  • conn.commit():PyMySQL 默认不会自动提交事务,增删改操作后需要手动提交。
  • 认证中间件:Flask 没有 Express 那种中间件链机制,但可以用装饰器来实现同样的功能。被 @login_required 装饰的路由会自动验证 Token。

十、用 Postman 测试

启动服务:

python app.py

用 Postman 测试三个接口:

  1. POST http://localhost:5000/api/register:Body → raw → JSON → {"name":"张三","email":"zhangsan@test.com","password":"123456"}。返回 201 和 Token。
  2. POST http://localhost:5000/api/login:同样的 JSON,返回 200 和 Token。
  3. GET http://localhost:5000/api/me:Headers → Authorization: Bearer 你的Token。返回当前用户信息。

十一、Flask 和 Express 的开发体验对比

维度 Express Flask
语言 JavaScript Python
路由定义 app.get('/path', fn) @app.route('/path') 装饰器
中间件机制 app.use(fn) 链式调用 装饰器 或 @app.before_request
异步支持 原生支持 async/await Flask 默认同步,异步需用 Quart 或 FastAPI
数据库生态 mysql2、sequelize、prisma PyMySQL、SQLAlchemy、Django ORM
适用场景 高并发 I/O、实时应用 传统 Web 应用、API 服务、数据接口

十二、本篇动手练习

练习 1:增加文章管理 API

app.py 中增加文章列表和文章详情的接口:GET /api/articlesGET /api/articles/<int:id>。数据从 MySQL 的 articles 表查询。

练习 2:增加发表文章接口

增加 POST /api/articles,需要登录认证。接收 titlecontent,存入数据库。

练习 3:用蓝图重构路由

Flask 的蓝图相当于 Express 的 express.Router()。把用户路由和文章路由分别拆分到 routes/users.pyroutes/articles.py 文件中,用蓝图注册到主应用。搜索“Flask Blueprint”了解语法。

十三、本篇小结

这一篇你用 Python 的 Flask 框架重写了后端服务:

  • Flask 基础Flask(__name__) 创建应用,@app.route('/') 装饰器定义路由,app.run() 启动服务。
  • 获取请求参数:路径参数 <int:id>、查询参数 request.args.get()、JSON 请求体 request.get_json()
  • 返回响应:返回字典自动转 JSON,return data, status_code 指定状态码。
  • CORS 跨域flask-cors 让前端能调用后端 API。
  • PyMySQL 操作数据库:连接 MySQL、执行 SQL、提交事务、DictCursor 返回字典格式。
  • JWT 认证jwt.encode() 生成 Token,自定义装饰器 @login_required 验证 Token。

如果你之前学过 Express,你会发现 Flask 写后端的感觉非常相似——接收请求、处理数据、返回响应,核心思路完全一致。掌握两种语言的后端能力,你就是真正的全栈开发者了。

下一篇预告

下一篇是这个系列的最后终结篇——《Python 零基础入门》的回顾与进阶路线。我们会总结你学过的所有内容,规划下一步的学习方向。

Python 零基础入门,每周更新。

© 版权声明
THE END
喜欢就支持一下吧
点赞8 分享
评论 抢沙发

请登录后发表评论

    暂无评论内容