aioshutil: Asynchronous shutil module

1.6 · active · verified Wed Apr 15

aioshutil is a Python library that provides asynchronous versions of functions from the standard `shutil` module. It achieves asynchronicity by running blocking I/O operations in a thread pool, thereby preventing the blocking of the asyncio event loop. The library is actively maintained, with the current version being 1.6.

Common errors

```
ModuleNotFoundError: No module named 'aioshutil'
```
cause The 'aioshutil' package is not installed in the Python environment.

fix Install the package using pip: 'pip install aioshutil'.
```
ImportError: cannot import name 'rmtree' from 'aioshutil'
```
cause The function 'rmtree' is not available in the 'aioshutil' module.

fix Ensure you are using the correct function names as provided by 'aioshutil'.
```
TypeError: 'coroutine' object is not callable
```
cause Attempting to call an asynchronous function from 'aioshutil' without using 'await'.

fix Use 'await' when calling asynchronous functions: 'await aioshutil.rmtree(path)'.
```
AttributeError: module 'aioshutil' has no attribute 'copytree'
```
cause The function 'copytree' is not available in the 'aioshutil' module.

fix Verify the available functions in 'aioshutil' and use the correct ones.
```
RuntimeError: This event loop is already running
```
cause Calling 'aioshutil' functions within an already running event loop, such as in Jupyter Notebook.

fix Use 'nest_asyncio' to allow nested use of 'asyncio' in environments like Jupyter Notebook.

Warnings

gotcha aioshutil performs its asynchronous operations by offloading blocking `shutil` calls to a thread pool (`loop.run_in_executor()`). This prevents blocking the asyncio event loop, but the underlying file operations are still blocking at the OS level. Users expecting true non-blocking I/O at a lower system level should be aware of this implementation detail.
Fix: Understand that 'asynchronous' in this context means non-blocking to the Python event loop, not necessarily non-blocking kernel I/O. Benchmark if extreme low-level I/O latency is critical.
gotcha When using `asyncio` objects (like `asyncio.Lock` or similar constructs) globally or outside of an `async` function before `asyncio.run()` is called, they might capture a stale reference to an implicit event loop. When `asyncio.run()` is invoked, it creates a new event loop, leading to potential runtime errors or unexpected behavior with the previously initialized objects. While primarily an `asyncio` footgun, it's relevant for any library interacting with the event loop like `aioshutil`.
Fix: Initialize `asyncio` primitives within an `async` function (e.g., inside `main()` or as part of an `__init__` method for async classes) after the event loop has been established by `asyncio.run()`. Python 3.10 and later mitigate this specific issue for `asyncio.Lock` by removing the `loop` parameter.

Install

pip install aioshutil Install with pip

Imports

copy
```
from aioshutil import copy
```
rmtree
```
from aioshutil import rmtree
```
move
```
from aioshutil import move
```

Quickstart

This quickstart demonstrates how to use `aioshutil` to perform common file operations like copying, moving, and removing directories asynchronously within an `asyncio` application. It mirrors the `shutil` API but with `await` calls.

import asyncio
import aioshutil
import os

async def main():
    # Ensure a directory exists for demonstration
    if not os.path.exists('test_dir_src'):
        os.makedirs('test_dir_src')
    with open('test_dir_src/test_file.txt', 'w') as f:
        f.write('Hello, aioshutil!')

    print('Copying file asynchronously...')
    await aioshutil.copy('test_dir_src/test_file.txt', 'test_dir_dest_copy.txt')
    print('File copied.')

    print('Moving file asynchronously...')
    await aioshutil.move('test_dir_src/test_file.txt', 'test_dir_dest_move.txt')
    print('File moved.')

    print('Removing directory asynchronously...')
    await aioshutil.rmtree('test_dir_src')
    print('Directory removed.')

    # Clean up created files/dirs if they exist from previous runs/failures
    if os.path.exists('test_dir_dest_copy.txt'):
        os.remove('test_dir_dest_copy.txt')
    if os.path.exists('test_dir_dest_move.txt'):
        os.remove('test_dir_dest_move.txt')

if __name__ == '__main__':
    asyncio.run(main())

view raw JSON →