aiohttp - 如何在类命名空间中保存持久会话

Question

我正在尝试在我的一个项目中使用 aiohttp，并努力弄清楚如何创建持久的aiohttp.ClientSession 对象。我浏览了官方的 aiohttp 文档，但在这方面没有找到帮助。

我浏览了其他在线论坛，发现自从 aiohttp 创建以来发生了很多变化。在 github 上的一些示例中，aiohttp 作者显示在 coroutine 函数（即 class Session: def __init__(self): self.session = aiohttp.ClientSession()）之外创建 ClientSession。我还发现不应该在协程之外创建ClientSession。

我尝试了以下方法：

class Session:
    def __init__(self):
        self._session = None

    async def create_session(self):
        self._session = aiohttp.ClientSession()

    async fetch(self, url):
        if self._session is None:
            await self.create_session()
        async with self._session.get(url) as resp:
            return await resp.text()

我收到很多关于 UnclosedSession 和连接器的警告。我也经常得到 SSLError。我还注意到，三分之二的呼叫挂起，我必须按 CTRL+C 才能杀死它。

使用requests，我可以简单地初始化__init__ 中的session 对象，但它不像aiohttp 那样简单。

如果我使用以下内容（这是我在各处看到的示例），我看不到任何问题，但不幸的是，在这里我最终会为每个请求创建 ClientSession。

def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()

我可以将aiohttp.ClientSession() 包装在另一个函数中并将其用作上下文管理器，但是每次调用包装函数时，我最终都会创建一个新的session 对象。我想弄清楚如何在类命名空间中保存 aiohttp.ClientSession 并重用它。

任何帮助将不胜感激。

Answer 1

这是工作示例：

from aiohttp import ClientSession, TCPConnector
import asyncio


class CS:

    _cs: ClientSession

    def __init__(self):
        self._cs = ClientSession(connector=TCPConnector(verify_ssl=False))

    async def get(self, url):
        async with self._cs.get(url) as resp:
            return await resp.text()

    async def close(self):
        await self._cs.close()


async def func():
    cs = CS()
    print(await cs.get('https://google.com'))
    await cs.close()  # you must close session


loop = asyncio.get_event_loop()
loop.run_until_complete(func())

Answer 2

你可以做到的。

我在写django程序的时候实现了一种session共享方式（使用asgi）。使用pid标记不同进程的session，方便django在不同进程之间调用。

经过实际测试，我可以直接调用共享会话。

• Django 3.2
• 独角兽

aiohttp.py

import os
import asyncio
import aiohttp
import logging

session_list = {}
logger = logging.getLogger(__name__)


class Req:

    @property
    def set_session(self):
        try:
            loop = asyncio.get_running_loop()
        except:
            loop = asyncio.get_event_loop()
            asyncio.set_event_loop(loop)
        session = aiohttp.ClientSession(loop=loop)
        session_list.update({os.getpid(): session})
        return session

    def __init__(self):
        if session_list.get(os.getpid()):
            self.session = session_list.get(os.getpid())
        else:
            self.session = self.set_session

    async def test(self):
        if session_list:
            session = session_list.get(os.getpid())
            if session and session.closed:
                session_list.pop(os.getpid())
                session = self.set_session
        else:
            session = self.set_session

        if not session or session.loop.is_running():
            session = self.set_session
            logger.warning("session abnormal")
        result = await session.get("http://httpbing.org/get")
        print(result.status)


req = Req()

views.py

from django.http import HttpResponse
from django.shortcuts import render  # noqa
from django.views.generic import View
from django.utils.decorators import classonlymethod

import asyncio


class TTT(View):

    @classonlymethod
    def as_view(cls, **initkwargs):
        view = super().as_view(**initkwargs)
        view._is_coroutine = asyncio.coroutines._is_coroutine
        return view

    async def get(self, request):
        await req.test()
        return HttpResponse("ok")