python-httpx的具体使用
作者:玉米丛里吃过亏 发布时间:2023-08-12 00:25:53
HTTPX是Python3的功能齐全的HTTP客户端,它提供同步和异步API,并支持HTTP/1.1和HTTP/2
安装
pip install httpx
创建请求
通过httpx库发出一个请求非常简单,如下:
import httpx
response = httpx.get('https://www.baidu.com/')
print(type(response), response) # <class 'httpx.Response'> <Response [200 OK]>
同样,我们再来发出一个POST请求:
response = httpx.post('http://localhost:5000/login', data={'username': 'httpx', 'password': '123456'})
PUT, DELETE, HEAD和OPTIONS请求都遵循相同的样式:
response = httpx.put('http://www.baidu.com/', data={key: value})
response = httpx.head('http://www.baidu.com/')
response = httpx.delete('http://www.baidu.com/')
response = httpx.options('http://www.baidu.com/')
自定义头部
要在传入请求中包含其他标头,请使用headers关键字参数:
header = {"user-agent": 'my_test/0001'}
response = httpx.get("https://api.github.com/events", headers=header)
超时时间
httpx设置默认的超时时间为5秒,超过此时间未响应将引发错误。我们可以通过timeout关键字参数来手动修改它:
response = httpx.get('http://localhost:5000/update', timeout=10)
你也可以将其设置为None完全禁用超时行为
response = httpx.get('http://localhost:5000/update', timeout=None)
超时又可以分为connect, read,write和pool超时。如果想详细设置,我们可以通过httpx.Timeout类来实现:
# 读取超时为10s,其他超时为5秒
timeout = httpx.Timeout(5, read=10)
response = httpx.get('http://localhost:5000/update', timeout=timeout)
SSL证书
通过httpx发出HTTPS请求时,需要验证所请求主机的身份。我们可以通过verify来指定我们存在的CA证书:
response = httpx.get('https://example.org', verify='../../client.pem')
或者你可以传递标准库ssl.SSLContext
import ssl
import httpx
context = ssl.create_default_context()
context.load_verify_locations(cafile='../../client.pem')
response = httpx.get('https://example.org', verify='../../client.pem')
又或者,你可以将verify设置为False禁用SSL验证:
response = httpx.get('https://example.org', verify=False)
认证
HTTPX支持Basic Auth和Digest Auth身份验证。要提供身份验证凭据,请将2个元组得纯文本str或bytes对象作为auth参数传递给请求函数:
response = httpx.get('https://example.com', auth=('my_user', 'password123'))
要提供Digest Auth身份验证得凭据,你需要Digest Auth使用纯文本用户名和密码作为参数实例化一个对象。然后可以将该对象作为auth参数传递给上述请求方法:
from httpx import DigestAuth
auth = DigestAuth('my_user', 'password123')
response = httpx.get('https://example.com', auth=auth)
httpx还提供了一个FunctionAuth类,允许我们传入一个Callable对象,该Callable接收request参数,并返回request。如下:
import httpx
from httpx._auth import FunctionAuth
def init_authorization(request):
request.headers['Authorization'] = 'Bearer 12334'
yield request
auth = FunctionAuth(init_authorization)
response = httpx.get('http://localhost:5000/home', auth=auth)
请求类型
Query Params
params = {"name":"zhangsan", "age":18}
response = httpx.get("https://www.baidu.com/s", params=params)
此时我们打印一下URL,发现该URL已经被正确编码:
print(response.url) # https://www.baidu.com/s?name=zhangsan&age=18
也可以传递一个列表数据进去:
params = {"name":"zhangsan", "favorite": ["football", "basketball"]}
response = httpx.get("https://www.baidu.com/s", params=params)
Form表单
通常情况下,你想要发送一些表单编码数据,就像HTML表单一样。要做到这一点,你只需要将字典传递给data关键字参数即可:
data = {'name': '张三'}
response = httpx.post('http://127.0.0.1:5000/test/post', data=data)
文件上传
你还可以使用HTTP分段编码上传文件
f = open('a.txt', 'rb')
files = {'file': f}
response = httpx.post('http://localhost:5000/post', files=files)
f.close()
JSON
如果你想要发送一个JSON数据,你可以通过将数据传递给json关键字参数即可:
response = httpx.post('http://127.0.0.1:5000/test/post', json={'name': '张三'})
二进制数据
对于其他编码,应使用content关键字参数,传递一个bytes类型数据
content = b'Hello World'
response = httpx.post('http://127.0.0.1:5000/test/post', content=content)
响应
响应类型
在上面的栗子可以知道,我们每次请求之后都会返回一个httpx.Response对象,我们可以从此对象中获取响应内容:
response = httpx.get("https://api.github.com/events")
print(type(response.text), response.text) # <class 'str'> [{"id":"14551634865","type":"PushEvent", ...}]
二进制响应
print(type(response.content), response.content) # <class 'bytes'> b'[{"id":"14551634865","type":"PushEvent", ...}]
JSON响应
print(type(response.json()), response.json()) # <class 'list'> [{'id': '14551634865', 'type': 'PushEvent', ...}]
流式响应
对于大型下载,你可能需要使用不将整个响应主体立即加载到内存中的流式响应。你可以流式传输响应的二进制内容:
for data in response.iter_bytes():
print(data)
流式传输响应的文本内容:
for text in response.iter_text():
print(text)
逐行流文本:
for line in response.iter_lines():
print(line)
原始字节:
for chunk in response.iter_raw():
print(chunk)
Cookie
如果响应包含了Cookie,你可以这样快速的访问它:
response = httpx.get('http://localhost:5050/get')
print(response.cookies['user'])
重定向历史
history响应的属性可用于检查任何后续的重定向。它包含遵循它们的顺序的所有重定向响应列表。例如GitHub将所有HTTP请求重定向到HTTPS:
response = httpx.get('http://github.com/')
print(response, response.url) # <Response [200 OK]> https://github.com/
print(response.history, response.history[0].url) # [<Response [301 Moved Permanently]>] http://github.com/
你还可以使用allow_redirects关键字参数来修改默认得重定向处理:
response = httpx.get('http://github.com/', allow_redirects=False)
print(response) # <Response [301 Moved Permanently]>
print(response.history) # []
httpx.Client
如果你会使用requests,那么可以使用httpx.Client代替requests.Session
with httpx.Client() as client:
response = client.get('http://localhost:5000/details')
另外,还可以使用.close()方法明确关闭连接池,而不会阻塞:
client = httpx.Client()
try:
response = client.get('http://localhost:5000/details')
finally:
client.close()
一旦你拥有了一个httpx.Client实例,那么你就可以通过调用.get()、.post()等方法发送请求。这些方法同样支持timeout、auth、headers等参数来满足我们的需求
合并/共享配置
httpx.Client还接收headers、cookie和params参数,对于同一组的请求操作,将共享同样的headers、cookie和params参数。如果请求方法中也包含了这些参数,那么它们将进行合并:
with httpx.Client(headers={'Token': '12345678'}, params={'page_size': 1, 'size': 20}) as client:
resp1 = client.get('http://localhost:5000/get', params={'search': 'laozhang'})
resp2 = client.post('http://localhost:5000/post')
如此,这两个请求的头部都将包含{'Token': '12345678'}。请求1的params将会合并,请求2将会使用{'page_size': 1, 'size': 20}查询参数
对于其他参数,如auth等,那么将会优先使用请求方法里面的auth
base_url
httpx.Client还允许使用base_url参数来设置基础URL,如下:
with httpx.Client(base_url='http://localhost:5000') as client:
response = client.get('/user/detail')
print(response.url) # http://localhost:5050/user/detail
limits
可以使用limits关键字参数来控制连接池的大小。它需要传递一个httpx.Limits类实例,httpx.Limits类接收以下两个参数:
max_keepalive: 最大活跃连接数,设置为None表示无限制。默认为10
max_connections:最大连接数,设置为None表示苏限制。默认为100
limits = httpx.Limits(max_keepalive=2, max_connections=5)
client = httpx.Client(limits=limits)
调用Python Web App
你可以配置httpx客户端以使用WSGI协议直接调用Python Web应用程序。这对于两个主要用例特别有用:
使用httpx的测试案例中的客户端
在测试期间或在dev/staging环境中模拟外部服务
import httpx
from flask import Flask
app = Flask(__name__)
@app.route("/home")
def home():
return 'Home Api Success'
with httpx.Client(app=app, base_url='http://testapi') as client:
response = client.get('/home')
print(response) # <Response [200 OK]>
print(response.text, response.url) # Home Api Success http://testapi/home
我们还可以通过使用WSGITransport来使用给定的客户端地址用于请求,如下:
transport = httpx.WSGITransport(app=app, remote_addr='1.2.3.4')
with httpx.Client(transport=transport, base_url='http://testapi') as client:
response = client.get('/home')
print(response) # <Response [200 OK]>
print(response.text, response.url) # Home Api Success http://testapi/home
如此,视图函数home中request.remote_addr将会是1.2.3.4
事件钩子
httpx允许你向客户端注册事件钩子,每次发生特定类型的事件时都会调用该钩子。httpx支持两个事件钩子:
request: 在即将发生请求时调用。为一个Callable列表,Callable接收httpx.Request实例参数
response: 响应返回后调用。为一个Callable列表,Callable接收httpx.Response实例参数
def _log_request(request):
print(type(request), request.url, request.method) # <class 'httpx.Request'> http://localhost:5000/hello GET
def _log_response(response):
print(type(response), response.url, response.text) # <class 'httpx.Response'> http://localhost:5000/hello Home Api Success
with httpx.Client(base_url='http://localhost:5000') as client:
client.get('/home')
代理
要将所有请求使用http://localhost:8030的代理,请将代理URL传递给Client:
with httpx.Client(proxies='http://localhost:8030') as client:
pass
对于更高级的使用,请使用dict。例如,要将HTTP和HTTPS请求路由到两个不同的代理: http://localhost:8030和http:localhost:8031:
proxies = {
'http://': 'http://localhost:8030',
'https://': 'http:localhost:8031'
}
with httpx.Client(proxies=proxies) as client:
pass
代理所有请求:
proxies = {
'all://': 'http://localhost:8030'
}
代理域名为“example.com”的所有请求:
proxies = {
'all://example.com': 'http://localhost:8030'
}
代理域名为“example.com”的所有HTTP请求:
proxies = {
'http://example.com': 'http://localhost:8030'
}
代理所有包含“example.com”的所有请求:
proxies = {
'all://*example.com': 'http://localhost:8030'
}
对于上面匹配,如果未匹配到将不使用代理。域名后面还可以添加端口号,用于更加严格的匹配。此外,我们还可以将匹配设置为None,用于排除,如下:
proxies = {
'all://': 'http://localhost:8030',
'all://example.com': None
}
即除使用“example.com”域名的路由,将使用“http://localhost:8030”代理
异步支持
HTTPX默认情况下提供标准的同步API,但是如果需要,还可以为你提供异步客户端的选项。要发出异步请求,你需要一个httpx.AsyncClient
import asyncio
import httpx
async def main():
async with httpx.AsyncClient() as client:
response = await client.get('https://example.org/')
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(main())
finally:
loop.close()
发出请求
请求方法都是异步的,因此你应该使用response = await client.get(...)样式对以下所有内容使用:
AsyncClient.get(url, ...)
AsyncClient.options(url, ...)
AsyncClient.head(url, ...)
AsyncClient.post(url, ...)
AsyncClient.put(url, ...)
AsyncClient.patch(url, ...)
AsyncClient.delete(url, ...)
AsyncClient.request(url, ...)
AsyncClient.send(url, ...)
流式响应
Response.aread()
Response.aiter_bytes()
Response.aiter_text()
Response.aiter_lines()
Response.aiter_raw()
来源:https://blog.csdn.net/y472360651/article/details/111313935
猜你喜欢
- 说明Python语言中列表(List)与其他语言的数组(Array)类似,是一种有序的集合数据结构,Python List可支持各种数据类型
- UEditor效果图一、简介UEditor是一个开源免费的编辑器,由百度web前端研发部开发所见即所得富文本web编辑器,具有轻量,可定制,
- 第一次见到《Web标准和SEO应用实践》是在我们UCD书友会《设计沟通十器》新书发布现场,机械工业出版社的朋友还带了很多相关新书过来,当时本
- 今天分享的这篇文章,文字不多,代码为主。绝对干货,童叟无欺,主要分享了提升 Python 性能的 20 个技巧,教你如何告别慢Python。
- 首先交代一下运行环境和工具版本:WIN10MINGW64ORACLEINSTANCCLIENT_18_3 x64Jetbrins Golan
- 在登陆PLSQL Developer时报ora-12513错误,需要修改本地数据库的监听配置文件,其过程相对简单。在Oracle安装目录中搜
- 在Linux下安装Oracle数据库是一件较为麻烦的事情。由于Linux的安装过程中可能会选择不同的安装包,会导致Oracle安装过程中缺失
- 信息图表设计(Inforgraphic Design),是信息设计(Information Design)学科的一个分支,它兴起于20世纪末
- 本文实例讲述了Python根据已知邻接矩阵绘制无向图操作。分享给大家供大家参考,具体如下:有六个点:[0,1,2,3,4,5,6],六个点之
- 一般情况下,使用pip命令安装即可:[root@dthost27 ~]# pip install mysql-python但是在实际工作环境
- numpy.insert()主要用于向矩阵中插入行或列。对于多维矩阵,可以沿任意一个轴插入元素。1. 参数说明numpy.insert(ar
- ImageField的使用笔记今天完善作业写的订单系统,主要是给每一个菜品增加图片,看起来美观一些,但是没想到这个小小的需求花了我一天时间,
- 1、概述 经常用到轮廓查找和多边形拟合等opencv操作,因此记录以备后续使用。本文代码中的阈值条件对图片没有实际意义,仅仅是为了测试。原图
- 看代码,再做解释<?php $array=array('a','b','c','
- 工欲善其事必先利其器,估计大部分的pythoner都是从python IDE开始了python之路。有很多的python IDE工具可以用,
- 关于Java和Mysql 8.0.18版本的连接方式,供大家参考,具体内容如下1.官网下载mysql-server.(Connector/J
- 本文实例讲述了PHP实现的线索二叉树及二叉树遍历方法。分享给大家供大家参考,具体如下:<?php require '
- 下面的demo是根据需求写的简单测试脚本#!/usr/bin/env python# coding: utf-8# 第一个列表为依赖组件和版
- 本文实例讲述了Python素数检测的方法。分享给大家供大家参考。具体如下:因子检测:检测因子,时间复杂度O(n^(1/2))def is_p
- 具体代码如下所示:#!/usr/bin/python# coding=utf-8from ftplib import FTPimport t