aclosing Class — langchain Architecture

Source Code

libs/core/langchain_core/utils/aiter.py lines 280–322

class aclosing(AbstractAsyncContextManager):  # noqa: N801
    """Async context manager to wrap an `AsyncGenerator` that has a `aclose()` method.

    Code like this:

    ```python
    async with aclosing(<module>.fetch(<arguments>)) as agen:
        <block>
    ```

    ...is equivalent to this:

    ```python
    agen = <module>.fetch(<arguments>)
    try:
        <block>
    finally:
        await agen.aclose()

    ```
    """

    def __init__(self, thing: AsyncGenerator[Any, Any] | AsyncIterator[Any]) -> None:
        """Create the context manager.

        Args:
            thing: The resource to wrap.
        """
        self.thing = thing

    @override
    async def __aenter__(self) -> AsyncGenerator[Any, Any] | AsyncIterator[Any]:
        return self.thing

    @override
    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        if hasattr(self.thing, "aclose"):
            await self.thing.aclose()