详解Python中httptools模块的使用

import httptools

# 第一个参数必须是 bytes 对象
url = httptools.parse_url(
    b"http://www.baidu.com"
)
# 返回一个 URL 对象
print(url.__class__)
"""
<class 'httptools.parser.parser.URL'>
"""

import httptools

# 第一个参数是 bytes 对象
url = b"http://satori:123456@www.baidu.com:80/s?wd=koishi#flag"
url_obj = httptools.parse_url(url)

print("协议:", url_obj.schema)
print("IP:", url_obj.host)
print("端口:", url_obj.port)
print("路径:", url_obj.path)
print("查询参数:", url_obj.query)
print("锚点:", url_obj.fragment)
print("用户信息:", url_obj.userinfo)
"""
协议: b'http'
IP: b'www.baidu.com'
端口: 80
路径: b'/s'
查询参数: b'wd=koishi'
锚点: b'flag'
用户信息: b'satori:123456'
"""

from pprint import pprint
import httptools

# 请求报文
request_payload = b"""POST /index?a=1 HTTP/1.1
Host: localhost:8080
Connection: keep-alive
Content-Length: 26
Cache-Control: max-age=0
Upgrade-Insecure-Requests: 1
Accept: text/html
Accept-Encoding: gzip, deflate, sdch
Cookie: _octo=GH1.1.1989111283.1493917476; logged_in=yes

{"name":"satori","age":17}"""

class Request:
    """
    将请求报文的解析结果封装成 Request 对象
    """
    def __init__(self):
        self.headers = {}
        self.body = b""
        self.path = None

    def on_url(self, path: bytes):
        self.path = path

    def on_header(self, name: bytes, value: bytes):
        self.headers[name] = value

    def on_body(self, body: bytes):
        self.body = body

# 实例化 Request 对象
request = Request()
# 将 request 作为参数传到 HttpRequestParser 中
parser = httptools.HttpRequestParser(request)
# 传入请求报文，进行解析
parser.feed_data(request_payload)

# 获取 HTTP 版本
print(parser.get_http_version())
"""
1.1
"""
# 是否是长链接（Connection 指定为 keep-alive）
print(parser.should_keep_alive())
"""
True
"""
# 获取请求方法
print(parser.get_method())
"""
b'POST'
"""
# 以上几个都是 HttpRequestParser 对象的方法

# 获取路径
print(request.path)
"""
b'/index?a=1'
"""

# 获取请求头
pprint(request.headers)
"""
{b'Accept': b'text/html',
 b'Accept-Encoding': b'gzip, deflate, sdch',
 b'Cache-Control': b'max-age=0',
 b'Connection': b'keep-alive',
 b'Content-Length': b'26',
 b'Cookie': b'_octo=GH1.1.1989111283.1493917476; logged_in=yes',
 b'Host': b'localhost:8080',
 b'Upgrade-Insecure-Requests': b'1'}
"""
# Cookie 也是请求头的一部分，但在解析的时候会单独拿出来
# 再解析成一个字典，然后通过 request.cookies 获取

# 获取请求体
print(request.body)
"""
b'{"name":"satori","age":17}'
"""

from pprint import pprint
import httptools

# 响应报文
response_payload = b"""HTTP/1.1 200 OK
Server: TornadoServer/6.1
Content-Type: text/html; charset=UTF-8
Date: Sun, 22 May 2022 17:54:11 GMT
Content-Length: 21

name: satori, age: 17"""

class Response:
    """
    将响应报文的解析结果封装成 Response 对象
    """
    def __init__(self):
        self.headers = {}
        self.body = b""
        self.status = b""

    def on_header(self, name: bytes, value: bytes):
        self.headers[name] = value

    def on_body(self, body: bytes):
        self.body = body

    def on_status(self, status: bytes):
        self.status = status

# 实例化 Response 对象
response = Response()
# 将 response 作为参数传到 HttpResponseParser 中
parser = httptools.HttpResponseParser(response)
# 传入响应报文，进行解析
parser.feed_data(response_payload)

# 获取 HTTP 版本
print(parser.get_http_version())
"""
1.1
"""
# 是否是长链接（不指定 Connection，默认为长连接）
print(parser.should_keep_alive())
"""
True
"""
# 获取状态码
print(parser.get_status_code())
"""
b'OK'
"""

# 获取状态码对应的描述
print(response.status)
"""
b'OK'
"""

# 获取响应头
pprint(response.headers)
"""
{b'Content-Length': b'21',
 b'Content-Type': b'text/html; charset=UTF-8',
 b'Date': b'Sun, 22 May 2022 17:54:11 GMT',
 b'Server': b'TornadoServer/6.1'}
"""

# 获取响应体
print(response.body)
"""
b'name: satori, age: 17'
"""