DOCSEARCHbyAlgolia
Submit Docs

aiofiles

repository·main·Indexed 25 days ago

https://github.com/tinche/aiofiles

File support for asyncio applications. aiofiles provides asynchronous file IO by delegating blocking local disk operations to a separate thread pool. It includes an asynchronous open() API, async access to standard streams (stdin, stdout, stderr), and the aiofiles.os and aiofiles.tempfile modules for asynchronous filesystem operations and temporary file management.

Tokens
2.7K
Snippets
5
Records
21
Agent score
84%

What's inside aiofiles

  2. Mock aiofiles for testing

    main

    To mock real file IO in tests, patch aiofiles.threadpool.sync_open. You must also register the return type with the aiofiles.threadpool.wrap dispatcher to ensure the mock is treated as an asynchronous object.

    import aiofiles
from unittest import mock

# Register the mock type with the dispatcher
aiofiles.threadpool.wrap.register(mock.MagicMock)(
    lambda *args, **kwargs: aiofiles.threadpool.AsyncBufferedIOBase(*args, **kwargs)
)

async def test_stuff():
    write_data = 'data'
    read_file_chunks = [b'file chunks 1', b'file chunks 2', b'']
    file_chunks_iter = iter(read_file_chunks)

    mock_file_stream = mock.MagicMock(
        read=lambda *args, **kwargs: next(file_chunks_iter)
    )

    with mock.patch('aiofiles.threadpool.sync_open', return_value=mock_file_stream) as mock_open:
        async with aiofiles.open('filename', 'w') as f:
            await f.write(write_data)
            assert await f.read() == b'file chunks 1'

        mock_file_stream.write.assert_called_once_with(write_data)

  4. Use aiofiles.open() for asynchronous file IO

    main

    Use aiofiles.open() to open files asynchronously. It mirrors the standard Python open() API but returns an asynchronous file object where IO methods are coroutines. You can optionally provide loop and executor arguments to control the event loop and thread pool used for delegating operations.

    import aiofiles

async with aiofiles.open('filename', mode='r') as f:
    contents = await f.read()
print(contents)

  5. Use aiofiles.tempfile for asynchronous temporary files

    main

    The aiofiles.tempfile module provides asynchronous interfaces for temporary file and directory management. It implements:

    • TemporaryFile
    • NamedTemporaryFile
    • SpooledTemporaryFile
    • TemporaryDirectory

    Results are wrapped in context managers compatible with async with and async for.

    import aiofiles.tempfile
import os

# Using NamedTemporaryFile
async with aiofiles.tempfile.NamedTemporaryFile('wb+') as f:
    await f.write(b'Line1\n Line2')
    await f.seek(0)
    async for line in f:
        print(line)

# Using TemporaryDirectory
async with aiofiles.tempfile.TemporaryDirectory() as d:
    filename = os.path.join(d, "file.ext")

  6. Use aiofiles.os for asynchronous filesystem operations

    main

    The aiofiles.os module provides coroutine versions of several os functions that delegate to an executor. Supported functions include:

    • File manipulation: stat, rename, renames, replace, remove, unlink, link, symlink, readlink
    • Directory operations: mkdir, makedirs, rmdir, removedirs, listdir, scandir
    • Path/Access: access, getcwd, path.abspath, path.exists, path.isfile, path.isdir, path.islink, path.ismount, path.getsize, path.getatime, path.getctime, path.samefile, path.sameopenfile
    • Data transfer: sendfile

  7. Access async stdin, stdout, and stderr

    main

    The following attributes provide asynchronous access to standard system streams and their underlying buffers:

    • aiofiles.stdin / aiofiles.stdin_bytes
    • aiofiles.stdout / aiofiles.stdout_bytes
    • aiofiles.stderr / aiofiles.stderr_bytes

  8. Access asynchronous standard streams (stdin, stdout, stderr)

    main

    The aiofiles package provides asynchronous access to standard input, output, and error streams. It offers both text-based and byte-based variants:

    • Text streams: stdin, stdout, stderr
    • Byte streams: stdin_bytes, stdout_bytes, stderr_bytes

  9. Create a named temporary file with NamedTemporaryFile

    main

    Use NamedTemporaryFile to asynchronously open a temporary file that has a visible name in the file system. It returns an AiofilesContextManager which can be used in an async with block.

    Parameters:

    • mode: File mode (default "w+b").
    • buffering: Buffering policy (default -1).
    • encoding: Encoding used if mode is text (default None).
    • newline: Newline character (default None).
    • suffix: Suffix to add to the filename.
    • prefix: Prefix to add to the filename.
    • dir: Directory where the file is created.
    • delete: Whether to delete the file on close (default True).
    • delete_on_close: (Python 3.12+) Whether to delete the file when closed (default True).
    • loop: The asyncio event loop (default None).
    • executor: The executor to use for running synchronous calls (default None).

  12. Use asynchronous os.path functions in aiofiles

    main

    The aiofiles.ospath module provides asynchronous versions of the standard os.path functions. These functions are wrapped to run in an executor, preventing them from blocking the asyncio event loop.

    Available functions include:

    • Path manipulation & info: abspath(path)
    • File statistics: getatime(path), getctime(path), getmtime(path), getsize(path)
    • File type checks: exists(path), isdir(path), isfile(path), islink(path), ismount(path)
    • File comparison: samefile(path1, path2), sameopenfile(f1, f2)