跳到主要内容

解析请求和处理响应

前言​

本章整理请求解析（URL 参数、查询参数、表单、JSON、文件、请求头、Cookie）、request 全局对象，以及规范 API 响应内容。

解析 URL 参数​

from markupsafe import escape


@app.route("/user/<username>")
def show_user_profile(username):
    return f"User {escape(username)}"


# 指定 url 参数为整数类型
@app.route("/post/<int:post_id>")
def show_post(post_id):
    return f"Post {post_id}"


# 指定 url 参数为路径
@app.route("/path/<path:subpath>")
def show_subpath(subpath):
    return f"Subpath {escape(subpath)}"

URL 中的参数类型：

参数类型描述
string字符串类型，可以接收除 / 以外的字符
int整型
float浮点类型
path类似 string，但可以接收 /
uuidUUID 类型
any备选值中的任何一个

any 用法：用户名只能是 zhangsan、lisi、wangwu 中的其中一个

@app.route("/user/<any(zhangsan,lisi,wangwu):name>")
def get_user(name):
    return f"username: {name}"

默认参数：比如默认分页为 1，让 URL 更简洁

@app.get("/posts/<int:post_id>")
@app.get("/posts/<int:post_id>/<int:page>")
def get_posts(post_id, page=1):
    return f"Post {post_id} page {page}"

查询参数​

from flask import request


# http://127.0.0.1:5000/query?username=zhangsan&group=student
@app.get("/query")
def query_args():
    username = request.args.get("username")
    group = request.args.get("group")
    return f"username: {username}, group: {group}"

表单参数​

username = request.form["username"]
password = request.form["password"]
提示

request.form 是 MultiDict，用 request.form.get("key") 更安全（不存在时返回 None 而不是抛异常）。

JSON 请求体​

from flask import Flask, jsonify, request

app = Flask(__name__)


@app.post("/json")
def post_json():
    req: dict = request.get_json(force=False, silent=False)
    username = req.get("username")
    password = req.get("password")
    return jsonify({"u": username, "p": password})


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=5000)

request.get_json() 参数：

  • silent=True：解析失败时返回 None 而不是抛异常。
  • force=True：即使 Content-Type 不是 application/json 也尝试解析。

文件上传​

from flask import request


@app.route("/upload", methods=["GET", "POST"])
def upload_file():
    if request.method == "POST":
        f = request.files["the_file"]
        f.save("/var/www/uploads/uploaded_file.txt")

请求头参数​

from flask import Flask, request

app = Flask(__name__)


@app.route("/headers")
def get_headers():
    user_agent = request.headers.get("User-Agent")
    content_type = request.headers.get("Content-Type")
    return f"User-Agent: {user_agent}, Content-Type: {content_type}"
from flask import Flask, request

app = Flask(__name__)


@app.route("/cookies")
def get_cookies():
    session_id = request.cookies.get("session_id")
    return f"Session ID: {session_id}"

request 对象​

在 Flask 项目中，获取客户端提交的数据可以通过全局线程安全对象 flask.request 实现。

常用属性：

属性说明
args解析后的客户端查询参数
query_string未解析的查询参数（bytes）
url客户端请求的完整 URL
base_url不含查询字符串的 URL
host_url类似 base_url，带 scheme 和 host
host客户端请求时用的域名
remote_addr客户端 IP
headers请求头
json客户端发送的 JSON 请求体
is_secure是否 HTTPS/WSS 请求
path请求 URL 的 path 部分
method请求方法
authorization请求头中的 Authorization

服务端示例：

from http import HTTPStatus

from flask import Flask, Response, request

app = Flask(__name__)


@app.post("/")
def index():
    print(f"args: {request.args}")
    print(f"query_string: {request.query_string}")
    print(f"form: {request.form}")
    print(f"url: {request.url}")
    print(f"base_url: {request.base_url}")
    print(f"path: {request.path}")
    print(f"method: {request.method}")
    print(f"headers: {request.headers}")
    print(f"cookies: {request.cookies}")
    print(f"host: {request.host}")
    print(f"host_url: {request.host_url}")
    print(f"remote_addr: {request.remote_addr}")
    print(f"user_agent: {request.user_agent}")
    print(f"is_secure: {request.is_secure}")
    print(f"json: {request.json}")
    print(f"auth: {request.authorization}")
    return Response(mimetype="text/plain", response="Hello World", status=HTTPStatus.OK)


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=8000, debug=False)

客户端示例：

import json

import requests

url = "http://127.0.0.1:8000"
req_body = {"k1": "v1"}
headers = {"Content-Type": "application/json"}
cookies = {"username": "zhangsan"}

resp = requests.post(
    url=f"{url}/?q1=s1", data=json.dumps(req_body), headers=headers, cookies=cookies
)
print(resp.status_code, resp.text)

规范 API 响应内容​

设计 API 返回内容时，通常需要与前端约定统一的响应体格式，方便反序列化和其它服务调用。常见格式：

{
    "code": 200,
    "data": {
        "content": "this is /a/1"
    },
    "msg": "success"
}
  • code：状态码。简单系统可直接用 HTTP 状态码；大型系统可设计自定义状态码，如：
from enum import Enum


class BizStatus(Enum):
    OK = 200
    BadRequestA1 = 4001  # 请求参数异常-A情况
    BadRequestA2 = 4002  # 请求参数异常-B情况
  • msg：对当前 code 的补充说明。
  • data：响应体内容。

方案一：封装响应类​

from http import HTTPStatus

from flask import Flask, jsonify, make_response, request

app = Flask(__name__)


class JsonResponse:
    def __init__(self, code: HTTPStatus = HTTPStatus.OK, msg: str = "success", data=None):
        self.code = code
        self.msg = msg
        self.data = data

    def response(self):
        resp = make_response(jsonify({
            "code": self.code.value,
            "msg": self.msg,
            "data": self.data,
        }), self.code.value)
        resp.headers["Content-Type"] = "application/json"
        return resp


@app.errorhandler(404)
def error_handler_not_found(error):
    return JsonResponse(
        code=HTTPStatus.NOT_FOUND,
        msg=f"{request.method} {request.path} Not Found",
    ).response()


@app.errorhandler(Exception)
def error_handler_generic(error):
    return JsonResponse(
        code=HTTPStatus.INTERNAL_SERVER_ERROR,
        msg=f"Internal Server Error. {request.method} {request.path}",
        data={"error": str(error)},
    ).response()


@app.get("/a/1")
def apitest_a1():
    return JsonResponse(
        code=HTTPStatus.OK, msg="success", data={"content": "this is /a/1"}
    ).response()

方案二：直接继承 Response​

import json
from datetime import datetime
from http import HTTPStatus

from flask import Flask, Response, request

app = Flask(__name__)


class CustomJSONEncoder(json.JSONEncoder):
    """datetime 不能直接序列化为 JSON，需要自定义编码器"""

    def default(self, o):
        if isinstance(o, datetime):
            return o.isoformat()
        return super().default(o)


class JsonResponse(Response):
    def __init__(
        self,
        data: dict | None = None,
        code: HTTPStatus = HTTPStatus.OK,
        msg: str = "success",
    ):
        headers = {"Content-Type": "application/json; charset=utf-8"}
        response = json.dumps(
            {"code": code.value, "msg": msg, "data": data},
            cls=CustomJSONEncoder,
        )
        super().__init__(response=response, status=code.value, headers=headers)


class Success(JsonResponse):
    def __init__(self, data: dict | None = None):
        msg = f"{request.method} {request.path} success"
        super().__init__(code=HTTPStatus.OK, msg=msg, data=data)


class Fail(JsonResponse):
    def __init__(self, data: dict | None = None):
        msg = f"Fail to {request.method} {request.path}"
        super().__init__(code=HTTPStatus.INTERNAL_SERVER_ERROR, msg=msg, data=data)


class ResourceNotFound(JsonResponse):
    def __init__(self, data: dict | None = None):
        msg = f"{request.method} {request.path} not found"
        super().__init__(code=HTTPStatus.NOT_FOUND, msg=msg, data=data)


@app.get("/")
def index():
    return Success({"now": datetime.now(), "sth": "hahahaha"})


@app.errorhandler(404)
def error_handler_notfound(error):
    return ResourceNotFound()


@app.errorhandler(Exception)
def error_handler_generic(error):
    return Fail()
提示

两种方案二选一即可。方案二继承 Response 后可直接作为视图返回值，配合 CustomJSONEncoder 处理 datetime 等特殊类型。更完整的独立实现见 补充：规范API响应内容。

参考​