正心全栈编程

读取请求数据 ​

文章服务需要根据搜索词查列表，也需要接收反馈。数据可能写在 URL、请求头或请求体里；读取位置不同，处理方式也不同。本专题沿着“查询文章 → 识别客户端 → 提交反馈”的过程使用 Request。

处理函数的 request 参数代表当前请求。query_params、path_params、headers 和 cookies 可以直接读取；读取 JSON 或表单请求体则需要 await。

示例代码见下文：request_data_demo.py 对应最终的 03_article_request_data.py，会依次检查各类请求数据。

读取查询参数与路径参数 ​

查询参数适合搜索、筛选与分页；路径参数用于标识具体文章。路由中的 {article_id:int} 会把路径片段转换为整数，处理函数再从 request.path_params 取值。

项目结构仍然保持简单：

text
p01c02/
└── 01_query_and_path.py

示例文件：01_query_and_path.py。

运行：uvicorn 01_query_and_path:app --reload。

py
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route


async def search(request):
    keyword = request.query_params.get("keyword", "")
    page = request.query_params.get("page", "1")

    return JSONResponse({
        "message": "文章搜索成功",
        "keyword": keyword,
        "page": page,
    })


async def article_detail(request):
    article_id = request.path_params["article_id"]

    return JSONResponse({
        "message": "文章详情读取成功",
        "article_id": article_id,
    })


routes = [
    Route("/search", search),
    Route("/articles/{article_id:int}", article_detail),
]

app = Starlette(routes=routes)

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("01_query_and_path:app", host='127.0.0.1', port=8000, reload=True)

这个例子做了两件事：

text
/search 读取查询参数
/articles/{article_id} 读取路径参数

关键代码：

python
request.query_params.get("keyword", "")
request.query_params.get("page", "1")

表示从 URL 查询参数中读取 keyword 和 page。

python
request.path_params["article_id"]

表示从路径参数中读取文章 ID。

启动查询示例 ​

执行：

bash
uvicorn 01_query_and_path:app --reload

区分查询参数与路径参数 ​

区分方法很简单：

text
写在 ? 后面的，是查询参数
写在路径里的，是路径参数

例如：

text
/search?keyword=python&page=1

这里 keyword 和 page 是查询参数。

text
/articles/1001

这里 1001 是路径参数。

URL 结构剖析与参数提取流程

读取请求头、Cookie 与 JSON ​

请求头通常描述客户端和内容格式，Cookie 保存客户端随请求带来的小段数据。JSON 放在请求体中；调用 await request.json() 后才能取得解析结果。需要原始字节时可用 await request.body()，本例只处理 JSON。

示例文件：02_headers_cookies_json.py。

：uvicorn 02_headers_cookies_json:app --reload。

py
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route


async def client_info(request):
    user_agent = request.headers.get("user-agent", "")
    accept = request.headers.get("accept", "")
    visit_token = request.cookies.get("visit_token", "")

    return JSONResponse({
        "user_agent": user_agent,
        "accept": accept,
        "visit_token": visit_token,
    })


async def feedback(request):
    request_data = await request.json()
    username = request_data.get("username")
    content = request_data.get("content")

    return JSONResponse({
        "message": "反馈提交成功",
        "username": username,
        "content": content,
    })


routes = [
    Route("/client-info", client_info),
    Route("/feedback", feedback, methods=["POST"]),
]

app = Starlette(routes=routes)

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("02_headers_cookies_json:app", host='127.0.0.1', port=8000, reload=True)

读取请求头 ​

关键代码：

python
request.headers.get("user-agent", "")

它会读取客户端请求头中的浏览器标识。

关键代码：

python
request.cookies.get("visit_token", "")

它会读取名为 visit_token 的 Cookie。

读取 JSON 请求体 ​

关键代码：

python
data = await request.json()

这里要特别注意：读取 JSON 请求体需要 await。

同步请求元数据 vs 异步 Body 数据流

验证 GET 接口 ​

访问：

text
http://127.0.0.1:8000/client-info

使用浏览器查看页面时，请求通常会自动带上 user-agent；演示脚本则显式设置该请求头。

验证 JSON 提交 ​

启动 02_headers_cookies_json.py 后，可用 requests.post(..., json=...) 向下面地址发送 POST 请求：

text
http://127.0.0.1:8000/feedback

脚本会打印状态码和解析后的响应体。

服务端会把 JSON 中的字段读出来，再返回给客户端。

读取表单数据 ​

表单和 JSON 都通过请求体提交，但编码方式不同。读取表单使用 await request.form()；运行本例前还需安装课程依赖中的 python-multipart。

很多浏览器表单提交不是 JSON，而是表单数据。

这时可以使用：

python
await request.form()

示例文件：03_article_request_data.py。

运行：uvicorn 03_article_request_data:app --reload。

python
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route


async def form_login(request):
    form = await request.form()
    username = form.get("username", "")
    password = form.get("password", "")

    return JSONResponse(
        {
            "message": "表单提交成功",
            "username": username,
            "password": password,
        }
    )


routes = [
    Route("/form-login", form_login, methods=["POST"]),
]

app = Starlette(routes=routes)

这里的流程是：

text
客户端提交表单
   ↓
request.form() 读取表单数据
   ↓
服务端取出 username 和 password
   ↓
返回 JSON 结果

验证表单接口 ​

可运行本专题的 request_data_demo.py，向下面地址发送 POST 请求：

text
http://127.0.0.1:8000/form-login

脚本会打印状态码和解析后的响应体。

JSON 与表单的区别 ​

这里可区分为：

text
JSON 更适合结构化接口数据
表单更常见于传统网页提交

两者都属于请求体，但读取方式不同：

python
await request.json()
await request.form()

请求体编码格式与解析管道对比（JSON vs Form）

文章查询与反馈案例 ​

最后把搜索、详情和反馈放到同一个应用中。

目标包括：

text
/search                  查询参数
/articles/{article_id:int}   路径参数
/client-info             请求头和 Cookie
/feedback                JSON 请求体
/form-login              表单数据

示例文件：03_article_request_data.py。

运行：uvicorn 03_article_request_data:app --reload。

py
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route


async def search_articles(request):
    keyword = request.query_params.get("keyword")
    page = request.query_params.get("page", "1")
    return JSONResponse({
        "message": "文章搜索成功",
        "keyword": keyword,
        "page": page,
    })


async def article_detail(request):
    article_id = request.path_params["article_id"]
    return JSONResponse({
        "message": "文章详情读取成功",
        "article_id": article_id,
    })


async def client_info(request):
    user_agent = request.headers.get("user-agent")
    accept = request.headers.get("accept")
    visit_token = request.cookies.get("visit_token")
    return JSONResponse({
        "user_agent": user_agent,
        "accept": accept,
        "visit_token": visit_token,
    })


async def feedback(request):
    request_data = await request.json()
    username = request_data.get("username")
    content = request_data.get("content")
    return JSONResponse({
        "message": "反馈提交成功",
        "username": username,
        "content": content,
    })


async def form_login(request):
    form_data = await request.form()
    username = form_data.get("username", "")
    password = form_data.get("password", "")
    return JSONResponse({
        "message": "表单提交成功",
        "username": username,
        "password": password,
    })


routes = [
    Route("/search", search_articles),
    Route("/articles/{article_id:int}", article_detail),
    Route("/client-info", client_info),
    Route("/feedback", feedback, methods=["POST"]),
    Route("/form-login", form_login, methods=["POST"]),
]

app = Starlette(routes=routes)

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("03_article_request_data:app", host='127.0.0.1', port=8000, reload=True)

案例需求 ​

这个案例把几种请求数据的读取方式放进同一个应用：

text
从 URL 读
从路径读
从请求头读
从 Cookie 读
从 JSON 读
从表单读

Starlette Request 数据读取全景矩阵

运行案例 ​

执行：

bash
uvicorn 03_article_request_data:app --reload

1、访问 /search

访问：

text
http://127.0.0.1:8000/search?keyword=python&page=1

2、访问 /articles/{article_id}

例如：

text
http://127.0.0.1:8000/articles/1001

3、访问 /client-info

可以直接访问：

text
http://127.0.0.1:8000/client-info

如果需要测试 Cookie，可以用工具手动带上 Cookie 再观察返回结果。

4、访问 /feedback

向下面地址发送 POST JSON：

text
http://127.0.0.1:8000/feedback

5、访问 /form-login

向下面地址发送表单 POST：

text
http://127.0.0.1:8000/form-login

案例源码 ​

py
import requests


BASE_URL = "http://127.0.0.1:8000"


def test_search():
    response = requests.get(
        f"{BASE_URL}/search",
        params={"keyword": "starlette", "page": 2},
        timeout=5,
    )
    print("GET /search", response.status_code, response.json())


def test_article_detail():
    response = requests.get(f"{BASE_URL}/articles/1", timeout=5)
    print("GET /articles/1", response.status_code, response.json())


def test_client_info():
    response = requests.get(
        f"{BASE_URL}/client-info",
        headers={"User-Agent": "Starlette-Course"},
        cookies={"visit_token": "abc123"},
        timeout=5,
    )
    print("GET /client-info", response.status_code, response.json())


def test_feedback():
    response = requests.post(
        f"{BASE_URL}/feedback",
        json={"username": "zhengxin", "content": "学习 Request"},
        timeout=5,
    )
    print("POST /feedback", response.status_code, response.json())


def test_form_login():
    response = requests.post(
        f"{BASE_URL}/form-login",
        data={"username": "zhengxin", "password": "123456"},
        timeout=5,
    )
    print("POST /form-login", response.status_code, response.json())


if __name__ == "__main__":
    # 启动 03_article_request_data.py 后依次观察五种读取方式。
    test_search()
    test_article_detail()
    test_client_info()
    test_feedback()
    test_form_login()