如何在多函数间共享aiohttp.ClientSession实现认证会话复用？

你当前代码的核心问题有两个：

1. Login()和do_something()各自通过async with Http()创建了独立的ClientSession，登录后的认证信息（比如Cookie）无法在两个会话间共享，导致后续请求用新会话访问受限页面时出现超时。
2. 两次调用asyncio.run()会创建独立的事件循环，前一次的会话资源会被彻底清理，无法跨循环复用。

aiohttp完全可以实现和requests一样的会话共享效果，核心是复用同一个ClientSession实例，并在同一个事件循环中完成所有相关操作。

方式一：直接传递ClientSession（类似requests的写法） #

这种方式最直观，和你给出的requests示例逻辑一致：

import asyncio
import aiohttp

async def login(session):
    url = "https://userlogin"  # 修正原代码的URL拼写错误
    data = {
        'email': 'mailm@mail.it',
        'pswd': '12345'
    }
    headers = {
        'Content-type': 'application/x-www-form-urlencoded; charset=UTF-8', 
        'Accept': 'application/json, text/javascript, */*; q=0.01',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:107.0) Gecko/20100101 Firefox/107.0',
    }
    try:
        async with session.post(url, data=data, headers=headers) as resp:
            resp.raise_for_status()
            return await resp.read()
    except Exception as e:
        print("登录异常: ", e)

async def do_something(session):
    url = "https://url_test.it?IsWork=0" 
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:107.0) Gecko/20100101 Firefox/107.0',
        'Content-type': 'application/x-www-form-urlencoded; charset=UTF-8'
    }
    try:
        async with session.get(url, headers=headers) as resp:
            resp.raise_for_status()
            return await resp.read()
    except Exception as e:
        print("操作异常: ", e)

async def main():
    # 创建一个全局复用的ClientSession，由async with自动管理生命周期
    async with aiohttp.ClientSession() as session:
        login_result = await login(session)
        print(login_result)
        # 异步场景用asyncio.sleep替代time.sleep，避免阻塞事件循环
        await asyncio.sleep(10)
        do_something_result = await do_something(session)
        print(do_something_result)

# 仅调用一次asyncio.run，确保所有操作在同一个事件循环中执行
asyncio.run(main())

方式二：调整封装的Http类支持外部传入会话 #

如果你想保留原有的Http封装，可以修改类逻辑，让它支持复用外部传入的ClientSession：

import asyncio
import aiohttp

class Http:
    def __init__(self, session=None):
        self._session = session

    async def __aenter__(self):
        # 仅在没有外部传入会话时，才新建会话
        if self._session is None:
            self._session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *err):
        # 仅清理自己创建的会话，外部传入的会话由调用方管理
        if self._session is not None and not self._session.closed:
            await self._session.close()
            self._session = None

    async def do_post(self, url, data, headers):
        async with self._session.post(url, data=data, headers=headers) as resp:
            resp.raise_for_status()
            return await resp.read()

    async def do_get(self, url, headers):
        async with self._session.get(url, headers=headers) as resp:
            resp.raise_for_status()
            return await resp.read()

async def login(http):
    url = "https://userlogin"
    data = {'email': 'mailm@mail.it', 'pswd': '12345'}
    headers = {
        'Content-type': 'application/x-www-form-urlencoded; charset=UTF-8', 
        'Accept': 'application/json, text/javascript, */*; q=0.01',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:107.0) Gecko/20100101 Firefox/107.0',
    }
    try:
        return await http.do_post(url, data=data, headers=headers)
    except Exception as e:
        print("登录异常: ", e)

async def do_something(http):
    url = "https://url_test.it?IsWork=0"
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:107.0) Gecko/20100101 Firefox/107.0',
        'Content-type': 'application/x-www-form-urlencoded; charset=UTF-8'
    }
    try:
        return await http.do_get(url, headers=headers)
    except Exception as e:
        print("操作异常: ", e)

async def main():
    async with aiohttp.ClientSession() as session:
        http = Http(session)  # 将全局会话传入Http实例
        login_result = await login(http)
        print(login_result)
        await asyncio.sleep(10)
        do_something_result = await do_something(http)
        print(do_something_result)

asyncio.run(main())

• 必须在同一个事件循环中完成登录和后续请求，避免多次调用asyncio.run()。
• 用asyncio.sleep()替代time.sleep()，后者会阻塞整个异步事件循环。
• ClientSession会自动维护Cookie等会话信息，只要复用同一个实例，就能保留认证状态。

内容的提问来源于stack exchange，提问作者Herojos