anyio-atexit 1.0.0

Small AnyIO utility for ensuring some task runs before the current async context finishes.

anyio-atexit

A small AnyIO utility for ensuring some tasks run before the current async context finishes.

This doesn't provide the same API as atexit, only shares a name for the sake of communicating intention.

Important Disclaimer:

It should be noted that this package is really intended for cases where you're porting or otherwise migrating an existing asyncio library / application to AnyIO. The idea of unmanaged and unaccountable "global" tasks is antithetical to structured concurrency principles, so should be considered carefully. In most cases, if you have the choice, refactoring using asynchronous context managers and task groups is likely a better long-term solution.

Installation

$ pdm add anyio-atexit

Optional extras:

• trio to use in a trio runtime.
• asyncio to use in an asyncio runtime. This installs asyncio-atexit.

Usage

A common use case is to ensure some free-floating asynchronous resource (like an IO client) gets gracefully closed before the loop exits in order to release a connection.

import anyio
from anyio_atexit import run_finally


class Foo:
    def __init__(self):
        # protect against users forgetting to close the client
        run_finally(self.disconnect)

    async def disconnect(self) -> None: ...


async def main():
    some_client = Foo()


anyio.run(main)

For anyio AsyncResources, you can also use the ensure_resource_closure function to automatically register the .aclose method, as well as get resource warnings and instantiation stack traces for unclosed resources.

You must implement a sync is_closed(self) -> bool method for the resource to be used with ensure_resource_closure.

Registering a finalizer for a resource does not make the async runtime aware of explicit closures. If you register a finalizer for a resource, you are also responsible for ensuring that the resource can be closed repeatedly without error for when the finalizer runs.

import anyio
from anyio_atexit import ensure_resource_closure


class Foo(AsyncResource):
    def __init__(self):
        self._closed = False
        ensure_resource_closure(self)

    def is_closed(self) -> bool:
        return self._closed

    async def aclose(self) -> None:
        if not self.is_closed():
            self._closed = True
            ...


async def main():
    some_client = Foo()


anyio.run(main)

License

This software is licensed under the Clear BSD License.