请教 aiohttp clientSession 认证过期带来的重定向问题

使用 aiohttp 的背景:

flask 项目需要访问阿里云上部署的 kubeflow api 服务, 这个 api 参数需要使用 query 传递, 因 query 构造过长报了 413,后考虑拆分 query 并发访问服务.

尝试 1 aiphttp + asyncio

大致代码抽象如下

import asyncio import json import re import functools from typing import Dict, List import aiohttp from loguru import logger def kubeflow_auth_with_async(func): """做 kubeflow 的 auth""" @functools.wraps(func) async def wrapper(*args, **kwargs): async with aiohttp.ClientSession() as session: # login payload = { "username": conf.USERNAME, "password": conf.PASSWORD, } await session.post(conf.AIX_AUTH_URL, data=payload) # get req # async with session.get(conf.PIPELINE_URL) as response: # text = await response.text() respOnse= await session.get(conf.PIPELINE_URL) text = await response.text() pattern = r"/dex/auth/aix\?req=" index_beg = re.search(pattern, text).span() index_end = text.find('"', index_beg[1]) req = text[index_beg[1] : index_end] params = {"req": req} # login kubeflow with aix await session.get(conf.DEX_AUTH_URL, params=params) return await func(session, *args, **kwargs) return wrapper async def async_get_runs( session: aiohttp.ClientSession, page_size=1, page_token=None, experiment_id=None, filters=None, sort_by="created_at desc", ): if not filters: filters = {} query = { "page_size": page_size, "page_token": page_token, "sort_by": sort_by, "resource_reference_key.type": "EXPERIMENT", "resource_reference_key.id": experiment_id, "filter": json.dumps(filters), } respOnse= await session.get( url=f"{conf.PIPELINE_URL}/apis/v1beta1/runs", params=query ) # async with session.get( # url=f"{conf.PIPELINE_URL}/apis/v1beta1/runs", params=query # ) as response: # response.raise_for_status() # return await response.json() response.raise_for_status() return await response.json() @kubeflow_auth_with_async async def gather_fetch_runs( session: aiohttp.ClientSession, runs_lst:List[str], ): exp_id = "xxxxx" tasks = [] for sub_runs in runs_lst: filters: dict = { "predicates": [ { "key": "id", "op": FILTER_OPERATIONS.IN.value, "string_values": {"values": sub_runs}, }, ] } tasks.append( async_get_runs( session, page_size=len(sub_runs), experiment_id=exp_id, filters=filters, ) ) return await asyncio.gather(*tasks) res = [] for r in asyncio.run(gather_fetch_runs(["xxx","xxx","xx"])): res.extend(r.get("runs", [])) 

请教 V 站大佬

出现的报错,不知道怎么贴图,手动概括一下异常

服务稳定运行一段时间后,会出现突然 500 刷新又可以坚挺一段时间 aiohttp.client_exceptions.ContentypTypeError 从报错信息上看像是 session 失效导致的,被 auth 重定向到了登录页. contentType 变成了 text/html 而请求的 url 重定向到了登录的 url,已经不是我传入的那个 因为没试过 async 函数的装饰器,不知道是不是这个问题, 还有就是 clientSession 的连接池是不是保存了 session 的状态. 装饰器每次都重新请求了.这个 session 按道理不应该还是一样的. 

尝试 2 gevent + request

也试了,没有这个问题,不过访问次数多起来之后容易莫名 GG,大概率是被自己家部署的 kubeflow 反爬了?