如何使用aiohttp重定向POST请求
aiohttp是一个基于Python的异步HTTP客户端和服务器框架，它支持异步请求和响应处理。使用aiohttp进行重定向POST请求可以通过以下步骤实现：
- 导入必要的模块和库：
import aiohttp
import asyncio- 创建一个异步函数来处理重定向POST请求：
async def redirect_post(url, data):
async with aiohttp.ClientSession() as session:
async with session.post(url, data=data, allow_redirects=True) as response:
return await response.text()- 调用上述函数并传入重定向目标URL和POST请求的数据：
url = "重定向目标URL"
data = {"key": "value"} # POST请求的数据
result = asyncio.run(redirect_post(url, data))
print(result)在上述代码中，我们使用了aiohttp的ClientSession来创建一个异步HTTP客户端会话，并使用post方法发送POST请求。allow_redirects=True参数允许重定向。最后，我们使用response.text()方法获取响应的文本内容。
需要注意的是，上述代码中的URL和数据仅作为示例，实际应用中需要根据具体情况进行修改。
推荐的腾讯云相关产品：腾讯云云服务器（CVM）和腾讯云函数（SCF）。
- 腾讯云云服务器（CVM）：提供可扩展的云服务器实例，可满足各种规模和需求的应用程序。您可以在CVM上部署和运行自己的应用程序，并使用aiohttp等工具进行开发和测试。
- 腾讯云函数（SCF）：是一种无服务器计算服务，可以让您在云端运行代码而无需管理服务器。您可以使用SCF来编写和运行异步函数，包括处理HTTP请求和响应。通过结合aiohttp和SCF，您可以实现高性能的异步HTTP请求处理。
更多关于腾讯云云服务器和腾讯云函数的详细信息，请访问以下链接：
- 腾讯云云服务器（CVM）：https://cloud.tencent.com/product/cvm
- 腾讯云函数（SCF）：https://cloud.tencent.com/product/scf
相关·内容
我正在使用python3和aiohttp http服务器来处理来自client.How的请求，我是否应该重定向包含正文数据的POST请求？ 我试过了 aiohttp.web.HTTPFound('/redirect') 它适用于GET请求，但是如果请求是POST呢？
浏览 51提问于2019-05-15得票数 0
1回答
我使用aiohttp客户端向URL发送请求并收集重定向URL。
在我的示例中，重定向URL中包含Unicode文本。我想要未经修改的。
例如，实际的重定向网址是example.com/förderprojekte-e-v，，但是aiohttp客户端自动编码它&返回me example.com/f\udcf6rderprojekte-e-v.
如何使aiohttp禁用重定向urls的自动编码。
对于请求模块，解决方案可以工作，但是我需要aiohttp的帮助。
我的代码：
async fetch(url):
#url = 'https://www.example.com/t
浏览 1提问于2021-08-08得票数 1
我试着让两个服务交流。第一个API向用户公开。第二个文件是隐藏的，可以处理文件。因此，第一个可以重定向请求。我想使用aiohttp创建post请求异步，但我面临这样的错误：“解析主体时出错了”。
要重新创建错误:让我们说这是服务器代码
from fastapi import FastAPI
from fastapi import UploadFile, File
app = FastAPI()
@app.post("/upload")
async def transcript_file(file: UploadFile = File(...)):
pass
这是客户
浏览 7提问于2021-05-31得票数 2
回答已采纳
我正在将一个使用请求的程序迁移到使用aiohttp，但是我已经找到了其中的一部分，我从使用aiohttp获得的响应与从请求得到的响应是不同的。
函数是这样的，并尝试将用户登录到网站。
async def LoginLN(usuario, contraseña, email, url, session):
s = session # aiohttp session
# some stuff and some other requests using the same aiohttp session to get the appropriate cookies for login
浏览 6提问于2022-09-25得票数 0
回答已采纳
1回答
为了简单起见，让我们使用一个简单的AIOHTTP应用程序
from aiohttp import web
async def handle(request):
name = request.match_info.get('name', "Anonymous")
text = "Hello, " + name
return web.Response(text=text)
app = web.Application()
app.add_routes([web.get('/', handle),
浏览 58提问于2022-07-29得票数 0
我正在执行aiohttp.ClientSession实例的request()，有时会引发asyncio.TimeoutError。我认为在这种情况下必须提出aiohttp.ServerTimeoutError，它是从asyncio.TimeoutError派生出来的，正如本文所说的：http://docs.aiohttp.org/en/stable/client_reference.html#aiohttp.ServerTimeoutError为什么会发生这种情况？也许是因为我使用的是旧版本的aiohttp? 2.3.8。 UPD这可能发生在非常简单的代码中，如下所示 async def e
浏览 11提问于2019-01-11得票数 4
回答已采纳
每当我对使用asyncio & aiohttp的client_exception.ServerDisconnectedError发出超过200个请求时，我就会得到一个aiohttp API。这似乎不是我的代码，因为它在较少的请求数下工作一致，但在较大的请求数上失败。试图理解这个错误是与aiohttp有关，还是与我的代码有关，还是与API端点本身有关？关于这个问题，网上似乎没有太多的信息。
Traceback (most recent call last):
File "C:/usr/PycharmProjects/api_framework/api_framework.
浏览 1提问于2018-07-09得票数 6
浏览器使用vue el上传组件上传文件，而aiohttp作为后端接收表单数据，然后保存it.but aiohttp request.multipart()始终为空，但request.post()将是可以的。
vue:
<el-upload class="image-uploader"
:data="dataObj"
drag
name="aaa"
:multiple="false"
:show-file-list="fa
浏览 0提问于2018-05-25得票数 2
跟踪器经常收到一个错误，我已经很长时间无法修复和复制了。
我的当事人就是这样
class Client:
_auth = (API_USER, API_PASSWORD)
def __init__(self):
self.auth = aiohttp.BasicAuth(self._auth[0], self._auth[1])
async def send(self, url, json, debug=True,):
try:
async with aiohttp.ClientSession(
浏览 1提问于2021-06-21得票数 3
回答已采纳
全!
我需要向web服务发出大约10,000个请求，并且我期望JSON响应。由于请求是相互独立的，所以我希望并行运行它们。我认为aiohttp可以帮我这个忙。我编写了以下代码：
import asyncio
import aiohttp
async def execute_module(session: aiohttp.ClientSession, module_id: str,
post_body: dict) -> dict:
headers = {
'Content-Type': r
浏览 8提问于2019-11-22得票数 8
我正在使用python aiohttp库编写一个web服务器。
如何设置cookie，然后在一个响应中将用户重定向到另一个页面？
可以使用aiohttp.web.HTTPSeeOther重定向用户，但我找不到将cookies附加到它的方法。
浏览 4提问于2017-02-15得票数 5
回答已采纳
2回答
我有一个脚本，它循环遍历一系列URL，根据返回的json数据提取项目位置。但是，脚本运行需要60分钟，其中55分钟(每个cprofile)用于等待json数据的加载。
我希望多线程一次运行多个POST请求，以加快速度，并且最初将URL范围分成两部分来完成。我陷入困境的地方是如何实现多线程或异步。
精简代码：
import asyncio
import aiohttp
# i am not recommend to use globals
results = dict()
url = "https://www.website.com/store/ajax/search"
que
浏览 6提问于2016-12-23得票数 0
1回答
我需要发送100多万个HTTP请求，到目前为止，我尝试过的每个选项都太慢了。我想我可以用aiohttp加快速度，但这似乎并不比请求更快。
我试着用python来做这件事，但我也对其他选项持开放态度。
这里是使用请求和aiohttp的代码，有任何加快进程的提示吗？
请求代码：
import requests
url = 'https://mysite.mysite:443/login'
users = [line.strip() for line in open("ids.txt", "r")]
try:
for user in us
浏览 1提问于2020-07-06得票数 0
我现在拥有的(Python3.4)：
r = yield from aiohttp.request('post', URL, params=None, data=values, headers=headers)
中的内容是：
conn = aiohttp.ProxyConnector(proxy="http://some.proxy.com")
r = await aiohttp.get('http://python.org', connector=conn)
那么，如何通过与aiohttp的代理连接发送带有标头的post请求呢？
谢谢。
浏览 4提问于2016-02-01得票数 4
回答已采纳
我正在尝试做一个功能，上传文件到我的谷歌驱动器。
下面的代码上传一个带有请求的文件。
import requests
import json
with open('storage.json', 'r') as file:
token = json.load(file)
token = token["access_token"]
url = "https://www.googleapis.com/upload/drive/v3/files"
file_metadata = {"name":
浏览 11提问于2022-05-27得票数 1
回答已采纳
1回答
我正在使用aiohttp请求从另一个web api下载一些内容，但是我收到了：
exception = TypeError('not a valid non-string sequence or mapping object',)
以下是我试图发送的数据。
data = "symbols=LGND-US&exprs=CS_EVENT_TYPE_CD_R(%27%27,%27now%27,%271D%27)"
如何解决这个问题？
我试过两种方法：
r = yield from aiohttp.get(url, params=data) # and
r =
浏览 3提问于2015-11-06得票数 3
回答已采纳
我希望使用aiohttp的多个请求。我是这样包装aiohttp的，我是这样测试的
我的代码
import asyncio
from aiohttp import ClientSession as AioClientSession
class ClientSession(AioClientSession):
async def _get(self, session, url, params=None, **kwargs):
async with session.get(url, params=params, **kwargs) as response:
浏览 0提问于2019-08-16得票数 1
1回答
我有一个接受post请求的web服务。post请求指定要在后台执行的特定作业，该作业修改用于稍后分析的数据库。请求的发送者不关心结果，只需要从web服务接收202确认。
到目前为止它是如何实现的: Flask Web服务将获得http请求，并向任务队列(rq工作进程)添加必要的参数，然后返回确认消息。一个单独的rq工作进程监听队列并处理作业。
我们现在已经切换到aiohttp，并意识到web服务现在可以通过使用aiohttp.ensure_future()方法在它自己的事件循环中调度实际的作业请求。
然而，这模糊了web服务器和任务队列之间的界限。从积极的方面来看，它消除了必须管理rq工作人员
浏览 4提问于2017-05-23得票数 2
描述了在aiohttp 中发送调用时的错误too many header bytes seen; overflow detected，并返回了错误too many header bytes seen; overflow detected。
来复制没有精确的代码来再现，因为它使用了专有的源代码。守则：
import asyncio
import aiohttp
loop = asyncio.get_event_loop()
url = "https://url.com"
bodies = [{"body": 1}, {"body": 2}, {
浏览 3提问于2021-04-09得票数 0
我有一些向某些API发出请求的代码。我想为所有的人建立普通的日志记录，我该如何设置呢？
假设我的代码如下所示
import aiohttp
import asyncio
async def fetch(client):
async with client.get('http://httpbin.org/get') as resp:
assert resp.status == 200
return await resp.text()
async def post_data(client):
async with client.p
浏览 3提问于2017-08-09得票数 12
回答已采纳
我正在使用aiohttp发出异步请求，我想测试我的代码。我想模拟aiohttp.ClientSession发送的请求。我正在寻找类似于为requests库处理模拟的方式。
我如何模拟aiohttp.ClientSession的反应？
# sample method
async def get_resource(self, session):
async with aiohttp.ClientSession() as session:
response = await self.session.get("some-external-api.com/resource
浏览 6提问于2019-08-28得票数 22
回答已采纳
1回答
我有test.py和AsyncioCurl.py文件。
我运行python .\Src\test.pt并给出了这个错误。
PS C:\Users\Thinkpad\Desktop\BilibiliHelper> python .\Src\test.py
2019-02-22 10:28:27,696 - C:\Users\Thinkpad\Desktop\BilibiliHelper\Src\Log.py[line:56] - DEBUG: GET: https://api.live.bilibili.com/room/v1/area/getRoomList?platform=web&a
浏览 3提问于2019-02-22得票数 0
我正在尝试做一个应用程序，可能会存活一天，一周或更长时间。在应用程序的生命周期内，它会向不同的API发出请求。其中一些API可能需要登录，因此我必须随时访问cookies。
所以我需要的是一个文件，不同的API可以使用而不会阻塞应用程序。
我对异步编程(asyncio/aiohttp)是个新手，我见过的例子展示了如何从一个url列表中发出很多请求，但这不是我需要的。
我的代码的问题是，要么得到ClientSession is closed错误，要么得到未关闭的ClientSession警告。
import asyncio # only here for debugging purposes
浏览 183提问于2018-09-08得票数 8
如何在aiohttp web服务器中创建http错误路由器？
from aiohttp import web
routes = list()
routes.append(web.route('POST', '/reg', handler))
例如，我如何创建：
routes.append(web.error_handler(404, handler404)
浏览 0提问于2021-03-08得票数 0
1回答
我对aiohttp和异步通信非常陌生，所以对我的无知表示歉意。我对文档的事件循环部分有困难，并且不认为下面的代码是异步执行的。我试图通过itertools获取两个列表的所有组合的输出，并将其发布到XML。在使用模块时，一个更完整的版本列出了requests，但是这并不理想，因为我需要一次发布1000+请求。下面是它现在的样子的一个示例：
import aiohttp
import asyncio
import itertools
skillid = ['7715','7735','7736','7737','7738&#
浏览 2提问于2021-02-23得票数 0
回答已采纳
当我使用网络套接字加入电子和Python时，我遇到了一些问题(主要是作为一种学习经验)：
使用电子Ok构建桌面应用程序
构建一个监视某些事情的Python程序-- Ok
使用socket.io 不工作的连接Python和电子
首先，我用中的“延迟”示例在python中测试了。当我看一看Chrome中的时，效果很好。
下一步是要求电子用mainWindow.loadURL('http://localhost:8080')在桌面应用程序中加载相同的url (我使用了官方electron-quick-start示例中的模板)。我只需在html中添加两行代码(如这里
浏览 2提问于2019-08-21得票数 3
回答已采纳
1回答
我是python的新手。我有一个异步函数，我正在尝试为它编写单元测试。 我纠结于如何等待函数返回响应，并使用它来执行下一行代码。 async def fetch_data(self, text, session):
response = None
payload = {"text": text.encode("ascii", "ignore").decode()}
try:
response = await session.post(url=self.embedder_endpoint, json=pay
浏览 48提问于2021-09-01得票数 1
1回答
说：
如果您想对aiohttp客户端使用超时，请使用标准异步方法: asyncio.wait_for(client.get(url)，10)
但这并不能处理DNS超时，我猜这是由操作系统处理的。另外，with aiohttp.Timeout不处理OS查找。
在上进行了讨论，没有最后的结论，而Saghul已经做出了，但是我不知道如何将它混合到aiohttp中，以及这是否允许asyncio.wait_for功能。
Testcase (在我的linux机器上花费20秒)：
async def fetch(url):
url = 'http://alicebluejeweler
浏览 4提问于2016-01-09得票数 4
我知道，但是答案并没有提到这里发生了什么。
在使用aiohttp时，我遇到了以下错误：
File "/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/selector_events.py", line 485, in sock_connect
return await fut
RuntimeError: await wasn't used with future
Unclosed client session
client_session: <aiohttp.
浏览 3提问于2021-04-11得票数 1
回答已采纳
1回答
给出了50k个网址的列表，我的任务是找出它们中的哪一个是可访问的。这样做的目的是向每个URL发送一个HEAD请求，并查看状态响应。据我所知，异步方法是可行的，目前我正在aiohttp中使用aiohttp。
我想出了下面的代码，但是速度非常糟糕。在我的10 10mbit连接上，1000个URL大约需要200秒。我不知道预期的速度，但我对Python的异步编程并不熟悉，所以我想我错了。正如您所看到的，我已经尝试将允许的同时连接的数量增加到1000 (从默认的100)和DNS解析保存在缓存中的持续时间；这两者都没有产生很大的效果。这个环境有Python3.6和aiohttp 3.5.4。
对与问题无
浏览 0提问于2019-03-19得票数 6
回答已采纳
一个很好的例子是站点登录: POST用户凭证和会话cookie将使用302设置( set - cookie )，但是在遵循重定向之后，ClientResponse.cookies、ClientResponse.History-1.cookie和ClientSession.cookie_jar都不包含会话cookie。这是aiohttp中的错误吗？注意，我确实在ClientResponse.history-1.headers中看到了cookie。
浏览 11提问于2019-02-20得票数 1
1回答
我有这样一个端点，它用csv返回一个文件：
# POST /export/
@asyncio.coroutine
def export(request):
post = yield from request.post()
if not post.get('passwd', None) == 'topsecret':
return web.Response(status=403)
csv = data.encode()
resp = web.StreamResponse(headers=aiohttp.MultiD
浏览 6提问于2015-12-04得票数 4
1回答
我正在编写应用程序，它每秒扫描目录，检查新文件，如果它们出现-通过POST请求发送它们并执行存档。假设目录中的文件数量可以从10到100 -我决定使用asyncio和aiohttp并发发送请求。
代码：
import os
import aiohttp
from aiohttp.client import ClientSession
BASE_DIR = '/path/to'
ARCHIVE_DIR = '/path/to/archive'
async def scan():
while True:
await asyncio.sle
浏览 14提问于2016-07-20得票数 6
回答已采纳
我正在实验limit和limit_per_host参数到aiohttp.connector.TCPConnector。
在下面的脚本中，我将connector = aiohttp.connector.TCPConnector(limit=25, limit_per_host=5)传递给aiohttp.ClientSession，然后向docs.aiohttp.org打开2个请求，向github.com打开3个请求。
session.request的结果是aiohttp.ClientResponse的一个实例，在本例中，我无意通过.close()或__aexit__对其调用.close()。我假
浏览 0提问于2018-12-28得票数 13
回答已采纳
1回答
每秒设置请求数
、、、
我正在用1000+请求编写一个API来获取数据，并且我希望限制每秒请求的数量。我试过：
conn = aiohttp.TCPConnector(limit_per_host=20)
and
conn = aiohttp.TCPConnector(limit=20)
但似乎不起作用
我的代码如下所示：
import logging
import asyncio
import aiohttp
logging.basicConfig(filename="logfilename.log", level=logging.INFO, format='%(asctime)s
浏览 7提问于2020-05-13得票数 1
回答已采纳
1回答
我使用aiohttp发送和接收响应，所以要记录某些日志，我需要向aiohttp的on_response_prepare方法发送一个带有request的自定义对象，那么我该如何做呢？
浏览 1提问于2018-03-13得票数 2
1回答
Asyncio中的缓存功能
、、、、
我正在尝试使用memcached从aiohttp缓存一个函数。缓存，我想作为一个装饰师。调用装饰师的示例如下所示：
@cache("main_page", expire=25)
@asyncio.coroutine
def page(request):
return aiohttp_jinja2.render_template('index', request, {} )
处理缓存的函数装饰器如下所示：
def cache_key(name, kwargs):
key = b'\xff'.join(bytes(k, 'ut
浏览 0提问于2015-10-02得票数 1
出于测试目的，尝试模拟aiohttp.ClientSession的响应
我的代码如下所示：
async def run_request(endpoint: str, **kwargs) -> dict:
async with aiohttp.ClientSession() as session:
async with session.post(endpoint, **kwargs) as response:
response_json = await response.json()
return {"res
浏览 13提问于2020-02-08得票数 1
我有一个异步函数从站点获取数据：
async def get_matches_info(url):
async with aiohttp.ClientSession() as session:
try:
async with session.get(url, proxy=proxy) as response:
...
...
...
...
except:
print('E
浏览 51提问于2019-07-16得票数 3
回答已采纳
我正在寻找解决方案如何设置域授权与aiohttp。有几个ldap库，但它们都阻塞了事件循环，再加上我对aiohttp的用户授权不是很清楚。正如我所看到的，我需要会话管理并将isLogdedIn=True存储在cookie文件中，在登录处理程序中检查每个路由->重定向的cookie，并检查每个模板中的关键字？它看起来非常不安全，会话可能会被窃取。
浏览 2提问于2015-08-06得票数 2
我想知道如何从aiohttp post方法获得当前的上传步骤。通常，我会使用get方法在一个循环中提取当前步骤，但是如果主机没有响应当前的上传步骤，这是行不通的。所以有可能得到当前的步骤吗？就像“从xx%上传几乎完成了”。我的意思是，等待上传完成是非常烦人的。
async def post_task():
archive = open("file")
session = aiohttp.ClientSession()
post = await session.post("upload_url", data=archive, ssl =
浏览 21提问于2022-04-11得票数 3
回答已采纳
我使用aiohttp.FormData()向Telegram chat发送文件 bytes_io = BytesIO(b"test file content 1")
file_content = bytes_io.getvalue()
file_name = "тест_1.txt"
async with aiohttp.ClientSession() as session:
post_file_data = aiohttp.FormData()
post_file_data.add_field(
"chat_id&
浏览 16提问于2021-09-21得票数 0
我正在尝试测试我的python 3.4.3http客户机或应用程序，它们将http请求发送到服务器。如果服务器因为某种原因而延迟响应，我希望请求不会被阻塞，因为asyncio+aiohttp应该提供异步调用：
def post(self):
print("received post")
print(self.request)
print("POST Body: %s" % str(self.request.body))
time.sleep(3)
self.write("blah")
self.fi
浏览 0提问于2018-01-30得票数 0
我想使用aiohttp发布多个帖子。而且，我需要与文件一起发布。但是，我的代码不工作这是我的代码
import aiohttp
file = open('file/path', 'rb')
async with aiohttp.request('post', url, files=file) as response:
return await response.text()
和request.FILES is None
这是trackback
def post(self, url: StrOrURL,
*
浏览 58提问于2019-08-19得票数 0
回答已采纳
我的问题是如何在aiohttp中做出正确的回应
官方的aiohttp文档为我们提供了创建异步查询的示例：
session = aiohttp.ClientSession()
async with session.get('http://httpbin.org/get') as resp:
print(resp.status)
print(await resp.text())
await session.close()
我不明白，为什么上下文管理器在这里。我所发现的只有__aexit__()方法等待resp.release()方法。但是这些文档也告诉我们，等待
浏览 2提问于2020-12-20得票数 7
回答已采纳
我想要错误的密码和错误的用户名的不同消息？？如何从响应中检测出用户名或密码是否错误？
auth = aiohttp.BasicAuth(username, password)
with aiohttp.ClientSession(loop=loop, auth=auth) as client:
client.post(Some_data) as response
浏览 2提问于2019-09-10得票数 0
1回答
给定具有类似于此视图的aiohttp.web应用程序：
async def hello(request):
return web.Response(body=b"Hello, world")
我在试着了解如何正确地测试它们。
在编写Django应用程序时，我通常使用Django自己的，并为aiohttp.web寻找类似的东西。我不确定这是不是正确的方法。
TL;DR:如何在单元测试中模拟对应用的请求？
浏览 2提问于2015-12-30得票数 6
回答已采纳
我有一个小型的web服务器：
# app.py
from typing import List
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
id: int
name: str
@app.post("/items")
async def items_list(items: List[Item]):
return items
if __name__ == "__main__":
浏览 10提问于2021-12-16得票数 0
我正在尝试使用基于证书的身份验证(cba)以及证书文件和密钥文件在服务器上进行身份验证。如何使用aiohttp实现这一点？使用用户名和密码的HTTP身份验证工作得很好，但我找不到cba的任何示例。
我尝试从aiohttp文档扩展这个示例。
async def main_ssl(loop):
sslcontext = ssl.create_default_context()
sslcontext.load_cert_chain(certfile=cert_file, keyfile=client_key)
conn = aiohttp.TCPConnector(ssl_c
浏览 10提问于2017-01-11得票数 2
回答已采纳
1回答
我有这样的方法
async def check_if_200(url):
async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(limit=0)) as session:
async with session.head(url) as response:
return response == 200
从200条协同线中调用，在一个无限循环中。有时，随机的，我得到：
File "/myProjectPath/myScript.py", line 71,
浏览 16提问于2022-03-23得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
云直播
对象存储活动推荐
腾讯云开发者
