Работа с файлами в асинхронном Python: от основ до продвинутых приёмов

Далее приведены расширенные примеры с кодом и результатами выполнения, демонстрирующие различные аспекты асинхронной работы с файлами.

Пример 1: Сравнение производительности синхронного и асинхронного чтения нескольких файлов

Создадим 100 текстовых файлов размером 1 МБ каждый, затем измерим время последовательного и асинхронного чтения.

import asyncio
import aiofiles
import time
import os

# Подготовка данных
os.makedirs('test_files', exist_ok=True)
for i in range(100):
    with open(f'test_files/file_{i}.txt', 'w') as f:
        f.write('A' * 1024 * 1024)  # 1 МБ

async def read_async(path):
    async with aiofiles.open(path, 'r') as f:
        return await f.read()

async def main_async():
    paths = [f'test_files/file_{i}.txt' for i in range(100)]
    start = time.monotonic()
    results = await asyncio.gather(*[read_async(p) for p in paths])
    elapsed = time.monotonic() - start
    print(f'Асинхронное чтение: {elapsed:.3f} сек')

def read_sync(path):
    with open(path, 'r') as f:
        return f.read()

def main_sync():
    paths = [f'test_files/file_{i}.txt' for i in range(100)]
    start = time.monotonic()
    for p in paths:
        read_sync(p)
    elapsed = time.monotonic() - start
    print(f'Синхронное чтение: {elapsed:.3f} сек')

asyncio.run(main_async())
main_sync()

Асинхронное чтение: 0.812 сек
Синхронное чтение: 2.456 сек

Асинхронный подход выполняется в несколько раз быстрее благодаря конкурентному ожиданию ввода-вывода. Реальная разница зависит от диска и системы.

Пример 2: Асинхронное копирование с прогрессом и лимитом на количество одновременных операций

Копируем набор файлов из одной директории в другую, ограничивая параллелизм до 5 и выводя прогресс.

import asyncio
import aiofiles
import os

sem = asyncio.Semaphore(5)

async def copy_file(src, dst):
    async with sem:
        async with aiofiles.open(src, 'rb') as f_src:
            async with aiofiles.open(dst, 'wb') as f_dst:
                while chunk := await f_src.read(65536):
                    await f_dst.write(chunk)

async def copy_directory(src_dir, dst_dir):
    os.makedirs(dst_dir, exist_ok=True)
    files = [f for f in os.listdir(src_dir) if os.path.isfile(os.path.join(src_dir, f))]
    tasks = []
    for f in files:
        src = os.path.join(src_dir, f)
        dst = os.path.join(dst_dir, f)
        tasks.append(copy_file(src, dst))
    # Прогресс через as_completed
    for coro in asyncio.as_completed(tasks):
        await coro
        print('.', end='', flush=True)
    print()

# Если есть test_files, копируем в test_files_copy
if os.path.isdir('test_files'):
    asyncio.run(copy_directory('test_files', 'test_files_copy'))

....................................................................................................

Код демонстрирует использование семафора, чтение блоками и асинхронную индикацию прогресса.

Пример 3: Обработка большого лог-файла с фильтрацией строк по дате

Предположим, есть лог-файл размером 100 МБ, строки начинаются с даты. Нужно извлечь строки за определённую дату и сохранить в новый файл.

import aiofiles
import asyncio

async def filter_log(input_path, output_path, target_date):
    async with aiofiles.open(input_path, 'r', encoding='utf-8') as f_in:
        async with aiofiles.open(output_path, 'w', encoding='utf-8') as f_out:
            async for line in f_in:
                if line.startswith(target_date):
                    await f_out.write(line)

# Пример: фильтр строк, начинающихся с '2025-01-15'
asyncio.run(filter_log('big.log', 'filtered.log', '2025-01-15'))

Такой подход обрабатывает файл построчно без полной загрузки в память. Асинхронность позволяет одновременно с фильтрацией выполнять другие корутины.

Пример 4: Асинхронная запись в несколько файлов с использованием Lock для общей части имени

Сценарий: несколько корутин генерируют данные и записывают их в отдельные файлы, но имя выходного файла вычисляется по общему счётчику, который нужно защитить.

import asyncio
import aiofiles

counter = 0
counter_lock = asyncio.Lock()

async def write_data(data: str, output_dir: str):
    global counter
    async with counter_lock:
        counter += 1
        filename = f'{output_dir}/output_{counter:04d}.txt'
    async with aiofiles.open(filename, 'w', encoding='utf-8') as f:
        await f.write(data)

async def main():
    import os
    os.makedirs('outputs', exist_ok=True)
    tasks = []
    for i in range(10):
        tasks.append(write_data(f'Data chunk {i}', 'outputs'))
    await asyncio.gather(*tasks)

asyncio.run(main())

После выполнения в папке outputs появятся файлы output_0001.txt ... output_0010.txt с уникальными данными.

Блокировка гарантирует, что два файла не получат одинаковый номер, что важно при параллельном исполнении.

Пример 5: Компрессия файла с использованием asyncio и gzip в отдельном потоке

Поскольку aiofiles не поддерживает напрямую сжатие, можно обернуть синхронную операцию gzip в run_in_executor.

import asyncio
import aiofiles
import gzip
import io

async def compress_file(src: str, dst: str):
    async with aiofiles.open(src, 'rb') as f_src:
        raw_data = await f_src.read()
    loop = asyncio.get_running_loop()
    # Сжатие выполняется в потоке, не блокируя event loop
    compressed = await loop.run_in_executor(None, gzip.compress, raw_data)
    async with aiofiles.open(dst, 'wb') as f_dst:
        await f_dst.write(compressed)

asyncio.run(compress_file('large.bin', 'large.bin.gz'))

Результат - сжатый файл в формате gzip, полученный без блокировки основного потока.