Как обрабатывать асинхронные добавления в файл CSV без повреждения

Если вы хотите сериализовать доступ к ресурсу с помощью экземпляра asyncio.Lock, тогда весь код доступа к ресурсу должен попытаться получить один и тот же экземпляр блокировки. Но у вас в коде:

        ...
        # Write the data
        for chunk in item:
            async with asyncio.Lock():
        ...

Следовательно, каждая задача постоянно создает новые экземпляры блокировки, которые не используются другими задачами.

Вторая проблема: у вас есть:

        ...
        if prefix == "file_one":
            # Shuffle the chunks again
            item = random.shuffle(item)
        ...

random.shuffle перемешивает последовательность на месте и возвращает None. Вместо этого у вас должно быть:

        ...
        if prefix == "file_one":
            # Shuffle the chunks again
            random.shuffle(item)
        ...

Я бы также рекомендовал, если у вас есть открытый файл f, то размер проще всего определить с помощью:

size = f.seek(0, 2)  # Seek to the end of file and return the offset

В следующем коде каждая задача открывает выходной CSV-файл один раз, но перед записью строки проверяет ее правильное расположение, переходя к текущему концу файла. Затем строка записывается, а выходной файл csv очищается перед освобождением управления:

import asyncio
import aiofiles
import aiofiles.os
import aiocsv
import uuid
import random
import json
from pathlib import Path
from datetime import datetime, timezone

async def write_csv(lock: asyncio.Lock, item: list,  load_id: str, prefix: str) -> None:

    Path("./test_files").mkdir(parents=True, exist_ok=True)
    file_path = Path("./test_files").joinpath(f"{prefix}_{load_id}.csv")

    async with aiofiles.open(file_path, mode = "a+", newline = "") as f:
        print(f"INFO: writing file: {Path(file_path).resolve()}")
        w: aiocsv.AsyncWriter = aiocsv.AsyncWriter(f)

        async with lock:
            # Asynchronously write to our file
                size = await f.seek(0, 2)  # Seek to end
                # If the file is empty, write the header
                if size == 0:
                    print("file was empty! writing header")

                    # Write the header
                    await w.writerow([
                        "response",
                        "load_id",
                        "last_updated_timestamp_utc"
                    ])
                    await f.flush()
            # Release the lock implicitly

        # do something special for specific file name
        # I am just trying to simulate more random data processing
        if prefix == "file_one":
            # Shuffle the chunks again
            random.shuffle(item)

        # Write the data
        for chunk in item:
            async with lock:
                size = await f.seek(0, 2) #  # Seek to end
                print(f"{file_path} file size: {size}")
                await w.writerow([
                    json.dumps(chunk),
                    load_id,
                    datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
                ])
                await f.flush()

async def main() -> None:

    # Create fake data
    items: list[str] = [["hello", "world"], ["asyncio", "test"]] * 500

    # Possible file prefixes
    prefixes: list[str] = ["file_one", "file_two"]

    tasks: list = []
    load_id = str(uuid.uuid4())
    lock = asyncio.Lock()
    for i in items:
        # Randomly assign which file we will write to
        task = asyncio.create_task(write_csv(lock, i, load_id, random.choice(prefixes)))
        tasks.append(task)

    errors = await asyncio.gather(*tasks, return_exceptions=True)
    # print(errors)

if __name__ == "__main__":
    loop = asyncio.new_event_loop()
    loop.run_until_complete(main())