参考链接:

  1. FastAPI-HTTPException
  2. 网络协议-七层、五层、四层协议概念
  3. HTTP协议入门-阮一峰

一、OSI七层网络模型

标准OSI七层网络模型:应用层(Application)、表示层(Presentation)、会话层(Session)、传输层(Transport)、网络层(Network)、数据链路层(Data Link)、物理层(Physical)。而实际使用提及的 TCP/IP 四层模型只不过是只使用了5 6 7层中的部分层+1234层而已。

下图可见:HTTP协议是基于 TCP 的应用层协议。HTTP负责定义传输内容的格式和规则,规定了请求和响应的结构以及各种请求方法的含义(如POST,GET)这样客户端和服务器就按照定好的规则组织和解析数据,HTTP 协议生成的请求和响应数据会被封装到TCP数据包中进行传输,TCP为HTTP的传输提供传输保障(提供面向连接的可靠字节流服务,确保数据完整有序到达对端)。

网络协议分层

二、HTTP/1.1 协议

a. 什么是 HTTP?

HTTP(HyperText Transfer Protocol)是Web浏览器和服务器间通信协议,当浏览器输入网址,调用API,本质上是在发送HTTP请求,服务器返回HTTP响应

HTTP/1.1 协议1997年发布,虽然在2015年发布了 HTTP/2(老设备不支持,二进制格式,抓包工具需要解码),但是 HTTP/1.1 仍然是最流行的版本(文本格式开发者友好,抓包工具直接可读)。

b. HTTP/1.1 协议原始报文格式详解

HTTP协议在应用层,所有HTTP报文就是字符串,使用 Postman 开源软件可以很方便的查看发送的HTTP报文长啥样,用curl命令怎么发,返回的响应长什么样子。

[!NOTE]

每个HTTP请求都会收到HTTP响应吗?

  • 协议语义上:每个成功被服务器接收并处理的请求都应返回一个响应(至少有状态行/状态码与响应头;特殊情况如 HEAD204 No Content304 Not Modified 没有响应体,但仍有响应)。
  • 实际网络层面:若在服务器处理前或处理过程中发生中间设备断开,超时,崩溃都可能导致客户端拿不到响应
  • HTTP 协议未规定响应的时间限制:只要连接保持打开,协议允许服务器过很久才返回响应。实际应用中超时来自各层的实现。
  • 所以,客户端应该做好异常处理,超时控制(自己定超时时间)
1)请求格式(Request):

包括:请求行,请求头,请求体。

GET /index.html HTTP/1.1  #  ① 请求行格式为 METHOD SP REQUEST-TARGET SP HTTP-VERSION CRLF [必须是 ASCII 文本(可视为字符串)]
Host: example.com         #  ② 请求头,指定服务器域名 [必须是 ASCII 文本(可视为字符串)]
User-Agent: curl/7.29.0
Accept: */*
name=Lili&age=25         #  ③ 可选请求体(Request Body)仅在 POST/PUT 中出现。[可以是任意二进制数据]

[!NOTE]

一、请求行,常见 METHOD:

  • GET:取资源
  • POST:提交数据
  • PUT / DELETE:REST 常用
  • HEAD:只请求头部
  • OPTIONS:查询服务器支持的方法

二、请求头

  • Host 必须存在(HTTP/1.1 新要求)

  • 键值对格式

  • 不区分大小写

  • 存在如下请求头可选键:

    Header 用途
    Host 必填,用于指定服务器域名,有了Host字段,就可以将请求发往同一台服务器上的不同网站,为虚拟主机的兴起打下了基础。
    User-Agent 浏览器/客户端标识
    Accept 可接受返回类型
    Content-Type 请求体类型
    Authorization Token、Basic Auth 等
    Connection: close 客户端在最后一个请求时,发送Connection: close,明确要求服务器关闭TCP连接。

三、请求体 body:

  • 可选请求体(Request Body)仅在 POST/PUT 方法出现,GET 请求通常没有请求体。
2)响应格式(Response):

包括:响应行,响应头,响应体。

HTTP/1.1 200 OK                         #  ① 响应行,含状态码
Content-Type: text/html; charset=UTF-8  #  ② 响应头,也是键值对格式
Content-Length: 1256
Connection: keep-alive
<html>...</html>                        #  ③ 响应体,任意数据(HTML、JSON、图片等)

[!NOTE]

一、响应行

  • 响应行中有如下状态码:

    类别 范围 含义 常见例子
    1xx 100-199 信息性(很少用) 100 Continue
    2xx 200-299 成功 200 OK, 201 Created
    3xx 300-399 重定向 301 Moved Permanently, 302 Found
    4xx 400-499 客户端 Client 错误(你搞错了) 400 Bad Request,
    401 Unauthorized,
    403 Forbidden,
    404 Not Found
    5xx 500-599 服务器 Server 错误(我搞错了) 500 Internal Server Error,
    502 Bad Gateway

    其中 FastAPI 默认如果在方法中通过return正常返回则响应的状态 码为200。但是你可以在想要返回异常时显式改变状态码:HTTPException(status_code=403, ...)

二、响应头

  • Headers 是键值对形式的元信息。用于告诉客户端如何处理响应。常见响应头:

    Header 作用
    Content-Type 来告诉接收方“如何解释 body”)
    如 字符串(JSON / HTML / XML)和非字符串如 图片(PNG/JPG),文件(PDF/ZIP),视频/音频流。Content-Type 的格式和值不是随便写的,而是全球统一的标准命名体系
    Content-Length 响应体的字节长度
    Cache-Control 是否缓存、缓存多久
    Set-Cookie 服务端设置 Cookie
    Server 服务端类型
    Connection keep-alive / close

    Content-Type 的常见标准 MIME 类型写法有:

    类别 MIME 类型 说明 是否文本
    JSON application/json JSON 数据
    HTML text/html HTML 文档
    PNG 图片 image/png PNG 格式图片
    二进制流 application/octet-stream 通用二进制(未知类型)

FastAPI 会自动设置合适的 Headers。比如当你返回 JSON 时,它会自动加:Content-Type: application/json。当你抛出 HTTPException,FastAPI 也会自动设置正确的 Content-Type 和状态码。

三、响应体 body:

符合响应头中Content-Type的类型的数据,客户端应当根据该类型解析。

❗️HTTP/1.1 连接特点(非常关键)默认是 Keep-Alive 长连接

  • 一个 TCP 连接可发送多个请求。但:不能多路复用必须按顺序返回(队头阻塞 Head-of-line blocking)

三、Python 实现 HTTP clinet 与 FastAPI server 通信

FastAPI server 一侧有 unvicorn 服务器实现回复,几乎不涉及HTTP的代码。

约定请求和响应JSON格式如下:

class RunScriptRequest(BaseModel):
    script_name: str = Field(..., description="逻辑脚本名")
    args: list[str] = Field(default_factory=list)
    timeout: int | None = Field(None, description="脚本超时时间(秒)")
class RunScriptResponse(BaseModel):
    status: str
    exit_code: int
    stdout: str
    stderr: str
    duration_ms: int

使用 Python requests 包的最小实现发送 HTTP POST 请求(只用 requests 发送 JSON、拿到响应、做基本校验与拆包,在拆包时用 Pydantic 严格校验响应结构):

import requests
from pydantic import BaseModel, Field, ValidationError
from typing import List, Optional

# ---- 契约模型(与你给出的一致) ----
class RunScriptRequest(BaseModel):
    script_name: str = Field(..., description="逻辑脚本名")
    args: List[str] = Field(default_factory=list)
    timeout: Optional[int] = Field(None, description="脚本超时时间(秒)")

class RunScriptResponse(BaseModel):
    status: str
    exit_code: int
    stdout: str
    stderr: str
    duration_ms: int


def run_script_with_validation(
    script_name: str,
    args: list[str] | None = None,
    timeout_seconds: int | None = None,
    request_timeout: float = 5.0,
) -> RunScriptResponse:
    """
    发送请求并用 Pydantic 严格校验响应。返回 RunScriptResponse 实例。
    """
    req = RunScriptRequest(script_name=script_name, args=args or [], timeout=timeout_seconds)

    try:
        resp = requests.post(
            url="https://your.api.host/run-script", # TODO: 替换为真实接口地址
            json=request_data, # 要发送的数据,自动将 Python 字典转换为 JSON 格式,自动设置Content-Type 为 application/josn
            timeout=timeout #超时设置(秒)
        )
        resp.raise_for_status()    # 非 2xx 会抛 HTTPError
    except requests.exceptions.Timeout:
        raise RuntimeError("请求超时")
    except requests.exceptions.ConnectionError:
    	print("连接错误")
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"请求失败:{e}")

    # 解析 JSON
    try:
        data = resp.json() # 将JSON格式转换为Python数据结构
    except ValueError:
        raise RuntimeError("响应内容不是合法 JSON")

    try:
        return RunScriptResponse.model_validate(data) # 验证格式是否符合预期
    except ValidationError as ve:
        # 打印/记录详细校验错误
        raise RuntimeError(f"响应结构不符合契约:{ve}")


if __name__ == "__main__":
    result = run_script_with_validation("daily_job", args=["--dry-run"], timeout_seconds=60)
    # 直接面向字段(有类型提示)
    print("状态:", result.status)
    print("退出码:", result.exit_code)
    print("耗时(ms):", result.duration_ms)
    print("标准输出:\n", result.stdout)
    print("标准错误:\n", result.stderr)

[!NOTE]

requests 基于 urllib3这个强大而灵活的客户端库,自动完成TCP建立,DNS解析,TLS握手与整数校验,无需手写无需自己写 socket/TCP,但如果希望复用 TCP 连接以提升性能,可使用 requests.Session()

上面的 requests.post 中的 json=request_data 为自动请求头的方法,何以替换成自动方法如下:

import json
headers = {'Content-Type': 'application/json'}
requests.post(url, data=json.dumps(data), headers=headers)
Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐