如何在异步循环关闭之前等待对象的 __del__ 完成？

Question

我有一个类，其中包含一个 aiohttp.ClientSession 对象。

通常在你使用时

async with aiohttp.ClientSession() as session:  
   # some code

会话将在会话的 __aexit__ 方法被调用后关闭。

我不能使用上下文管理器，因为我想在对象的整个生命周期内保持会话持久。

这行得通：

import asyncio
import aiohttp

class MyAPI:
    def __init__(self):
        self.session = aiohttp.ClientSession()

    def __del__(self):
        # Close connection when this object is destroyed
        print('In __del__ now')
        asyncio.shield(self.session.__aexit__(None, None, None))



async def main():
    api = MyAPI()

asyncio.run(main())

但是，如果在某些地方引发了异常，则事件循环会在 __aexit__ 方法完成之前关闭。 我该如何克服这个问题？

堆栈跟踪：

Traceback (most recent call last):
  File "/home/ron/.PyCharm2018.3/config/scratches/async.py", line 19, in <module>
    asyncio.run(main())
  File "/usr/local/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 568, in run_until_complete
    return future.result()
  File "/home/ron/.PyCharm2018.3/config/scratches/async.py", line 17, in main
    raise ValueError
ValueError
In __del__ now
Exception ignored in: <function MyAPI.__del__ at 0x7f49982c0e18>
Traceback (most recent call last):
  File "/home/ron/.PyCharm2018.3/config/scratches/async.py", line 11, in __del__
  File "/usr/local/lib/python3.7/asyncio/tasks.py", line 765, in shield
  File "/usr/local/lib/python3.7/asyncio/tasks.py", line 576, in ensure_future
  File "/usr/local/lib/python3.7/asyncio/events.py", line 644, in get_event_loop
RuntimeError: There is no current event loop in thread 'MainThread'.
sys:1: RuntimeWarning: coroutine 'ClientSession.__aexit__' was never awaited
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7f49982c2e10>

Answer 1

不要使用__del__ 挂钩来清理异步资源。你根本不能指望它被调用，更不用说控制它什么时候被使用或者异步循环在那个时候是否仍然可用。你真的想明确地处理这个。

要么使 API 成为异步上下文管理器，要么在退出时显式清理资源，例如使用 finally 处理程序； with 和 async with 语句基本上旨在封装传统上在 finally 块中处理的资源清理。

我会在这里将API 实例设为上下文管理器：

class MyAPI:
    def __init__(self):
        self.session = aiohttp.ClientSession()

    async def __aenter__(self):
        return self

    async def __aexit__(self, *excinfo):
        await self.session.close()

请注意，ClientSession.__aexit__() 真正所做的只是在 self.close() 上等待，所以上面直接指向那个协程。

然后在你的主循环中使用它：

async def main():
    async with MyAPI() as api:
        pass

另一种选择是将您自己的会话对象提供给MyAPI 实例，并在完成后自行负责关闭它：

class MyAPI:
    def __init__(self, session):
        self.session = session

async def main():
    session = aiohttp.ClientSession()
    try:
        api = MyAPI(session)
        # do things with the API
    finally:
        await session.close()

Answer 2

正如@Martijn Pieters 所说，您不能强制事件循环等待对象的__del__ 析构函数调用。但是，您仍然可以使用__del__ 析构函数来关闭异步资源，方法是首先检查循环是否正在运行，如果不是，则启动一个新循环。例如，asyncio Redis 模块使用这种技术when destructing its Client class。对于您的代码，具体而言，析构函数如下：

import asyncio
import aiohttp


class MyAPI:

    def __init__(self):
        self.session = aiohttp.ClientSession()

    def __del__(self):
        # Close connection when this object is destroyed
        try:
            loop = asyncio.get_event_loop()
            if loop.is_running():
                loop.create_task(self.session.close())
            else:
                loop.run_until_complete(self.session.close())
        except Exception:
            pass

Answer 3

谢谢你，@alan。 我使用了您的示例并为其添加了一些类型。我在课堂上使用pyppeteer。不是 100% 确定它是否正确，但至少不会再引发关于运行循环的异常，并且它作为 __del__ 的一部分执行。我现在使用它作为包装函数将我的异步代码转换为同步代码。这有点难看，但它有效。 我现在可以在对象被销毁时安全地关闭浏览器实例。

我输入的例子

from asyncio import get_event_loop
from typing import TypeVar, Callable, Coroutine, Any

ReturnType = TypeVar("ReturnType")


def async_to_sync(callable_function: Callable[[], Coroutine[Any, Any, ReturnType]]) -> ReturnType:
    loop = get_event_loop()
    if loop.is_running():
        return loop.create_task(callable_function())
    else:
        return loop.run_until_complete(callable_function())

Answer 4

我在写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")