1 from __future__ import annotations 2 3 from abc import ABCMeta, abstractmethod 4 from types import TracebackType 5 from typing import TypeVar 6 7 T = TypeVar("T") 8 9 10 class AsyncResource(metaclass=ABCMeta): 11 """ 12 Abstract base class for all closeable asynchronous resources. 13 14 Works as an asynchronous context manager which returns the instance itself on enter, 15 and calls :meth:`aclose` on exit. 16 """ 17 18 __slots__ = () 19 20 async def __aenter__(self: T) -> T: 21 return self 22 23 async def __aexit__( 24 self, 25 exc_type: type[BaseException] | None, 26 exc_val: BaseException | None, 27 exc_tb: TracebackType | None, 28 ) -> None: 29 await self.aclose() 30 31 @abstractmethod 32 async def aclose(self) -> None: 33 """Close the resource.""" 34