asyncio-atexit

1.0.1 · active · verified Wed Apr 15

asyncio-atexit is a Python library that provides `atexit`-like functionality for `asyncio` event loops. It allows users to register coroutines or synchronous functions to be executed when the current `asyncio` event loop closes, rather than when the Python interpreter exits. The current version is 1.0.1, and it is a small utility that is updated infrequently but remains active and functional for its specific purpose.

Warnings

gotcha Callbacks registered with `asyncio-atexit` are called without arguments. To pass arguments to your cleanup functions, use `functools.partial` to wrap your function with its arguments.
Fix: Wrap your target function with `functools.partial` before registering it, e.g., `asyncio_atexit.register(functools.partial(my_cleanup_func, arg1, kwarg='value'))`.
gotcha Callbacks are only invoked when the associated `asyncio` event loop is explicitly closed. If your application's `asyncio` event loop is never closed (e.g., in long-running services or certain environments like `nest_asyncio` where loops are reused), registered cleanup functions may not execute.
Fix: Ensure that `loop.close()` is called on the event loop where callbacks were registered. If using `asyncio.run()`, ensure it completes normally, as it typically handles loop closure. In complex scenarios or with libraries like `nest_asyncio`, you might need to manage loop lifecycle more explicitly.
gotcha `asyncio-atexit.register()` must be called from within a running `asyncio` event loop. Attempting to register a callback without an active loop will likely result in an error or unexpected behavior.
Fix: Ensure that `asyncio_atexit.register()` calls are made after the event loop has started, typically within an `async` function executed by `asyncio.run()` or `loop.run_until_complete()`.
breaking Using `asyncio-atexit` with `nest-asyncio` can be problematic, particularly on Python 3.14+. `nest_asyncio` alters the `asyncio` loop's closing behavior, often leaving the loop open and reused rather than truly closing it. This prevents `asyncio-atexit`'s hooks from firing, as they depend on the loop's `close()` method being called. Python 3.14+ further breaks `nest_asyncio` workarounds for nested loops.
Fix: Avoid using `asyncio-atexit` with `nest-asyncio` if loop-closing guarantees are critical. If `nest-asyncio` is indispensable, ensure manual `loop.close()` calls are made when the outermost logical 'event loop' concludes, or reconsider cleanup strategies that don't rely on `asyncio-atexit`'s specific hook.

Install

pip install asyncio-atexit Install stable version

Imports

register
```
from asyncio_atexit import register
```
unregister
```
from asyncio_atexit import unregister
```

Quickstart

Registers an asynchronous function and a synchronous function to be called when the asyncio event loop shuts down. This example demonstrates using `functools.partial` to pass arguments to the cleanup functions, as `asyncio-atexit` callbacks are invoked without arguments by default. The `asyncio.run` function typically manages loop closure, but an explicit `loop.close()` is added in the `finally` block to guarantee cleanup function execution in all environments for demonstration purposes.

import asyncio
import asyncio_atexit
import functools

async def async_cleanup_task(resource_name):
    print(f"Async cleanup for {resource_name} started...")
    await asyncio.sleep(0.1) # Simulate async work
    print(f"Async cleanup for {resource_name} finished.")

def sync_cleanup_task(message):
    print(f"Sync cleanup: {message}")

async def main():
    print("Main application started.")
    # Register an async cleanup task
    asyncio_atexit.register(functools.partial(async_cleanup_task, "Database Connection"))
    # Register a synchronous cleanup task
    asyncio_atexit.register(functools.partial(sync_cleanup_task, "Closing log file."))
    print("Cleanup tasks registered.")
    await asyncio.sleep(0.5) # Simulate main application work
    print("Main application finishing.")

if __name__ == '__main__':
    try:
        asyncio.run(main())
    finally:
        # In some scenarios, especially with older asyncio.run or specific policies,
        # you might explicitly close the loop if it wasn't handled by asyncio.run.
        # asyncio-atexit hooks only run if the event loop is actually closed.
        # This ensures the example triggers the atexit hooks.
        loop = asyncio.get_event_loop()
        if not loop.is_closed():
            loop.close()

view raw JSON →