Связь между очередями в Asyncio не работает должным образом

Мой первый комментарий заключается в том, что ваш код гораздо сложнее, чем мне нужно. Но настоящая проблема заключается в том, что вам нужно выполнить N выборок (где N — длина вашего списка items) и M задач (где M в настоящее время равно 5), выполняющих эти N выборок одновременно, но нельзя предположить, что эти задачи будут выполнены за один раз. заказ, соответствующий вашему items списку.

Я считаю, что самое простое решение — предварительно выделить список results длиной N, и каждому запросу на выборку передается этот список и индекс, указывающий, куда должен идти результат в этом списке. Вы не можете начать запись файла CSV, пока не будут завершены все выборки, если вы хотите, чтобы строки файла соответствовали входному списку items. Вам нужна только одна очередь!

Для демонстрационных целей я переименовал items в более описательный data_list и инициализировал его как:

data_list: list[str] = [[f"hello{i}", f"world{i}"] for i in range(10)]

Я также модифицировал fetch, чтобы имитировать получение данных и простой возврат req[data]. Следовательно, когда программа завершит работу, содержимое файла file.csv должно быть:

Я также переименовал некоторые ваши функции и переменные в более описательные:

import asyncio
import aiohttp
import aiofiles
import aiocsv
import json

N_REQUEST_TASKS = 5

async def fetch(req: dict, results: list, idx: int) -> dict:
    # Make the request

    # For demo purposes:
    import random

    await asyncio.sleep(random.random())
    result = req['data']
    print(f"INFO returning {result} at index {idx}")
    results[idx] = result
    return

    async with aiohttp.ClientSession() as session:
        try:
            async with session.request("POST", url=req["url"], data=json.dumps(req["data"]), headers=req["headers"]) as response:
                result = await response.json()
                response.raise_for_status()
                print(f"INFO: response status was: {response.status}")

                # Put response into queue to be written to file
        except Exception as err:
            print(f"ERROR: error making request: {err}")
            result = err
        finally:
            print(f"INFO returning {result} at index {idx}")
            results[idx] = result


async def create_requests(data_list: list,  results: list, request_q: asyncio.Queue) -> None:
    # Read tasks from source of data
    for idx, data in enumerate(data_list):
        # Put a request task into the queue
        req: dict = {
            "headers": {"Accept": "application/json"},
            "url": "https://httpbin.org/post",
            "data": data
        }
        await request_q.put(
            (req, results, idx)
        )

    for _ in range(N_REQUEST_TASKS):
        # One sentinel for each request task:
        await request_q.put(None)


async def request_task(request_q: asyncio.Queue) -> None:
    while True:
        # Retrieve necessary data to make request
        request = await request_q.get()

        # Sentinel?
        if not request:
            print("INFO: received sentinel from request_q")
            break

        # Make the request which will put data into the response queue
        # Unpack:
        req, results, idx = request
        print(f"INFO: request in request_task: {req['data']}")
        await fetch(req, results, idx)


async def writer(results: list) -> None:
    async with aiofiles.open("file.csv", mode = "w", newline='') as f:
        w = aiocsv.AsyncWriter(f)

        await w.writerow([
            "status",
            "data",
        ])

        for result in results:
            print(f"INFO: data in write_task: {result}")
            if isinstance(result, Exception):
                continue

            # Write the data from the response
            await w.writerow([
                "200",
                json.dumps(result)
            ])
            await f.flush()

async def main() -> None:

    # Create fake data to POST
    data_list: list[str] = [[f"hello{i}", f"world{i}"] for i in range(10)]

    # Preallocate results list so that results will be in correct order
    results = [None] * len(data_list)

    # Request queue
    request_q = asyncio.Queue()

    tasks = []

    # one producer
    tasks.append(
        asyncio.create_task(
            create_requests(data_list, results, request_q)
        )
    )

    # N_REQUEST_TASKS consumers
    for _ in  range(N_REQUEST_TASKS):
        tasks.append(
            asyncio.create_task(
                request_task(request_q)
            )
        )

    await asyncio.gather(*tasks)
    print(f"INFO: Results have been produced")

    await writer(results)
    print("INFO: writer has completed! ")

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

Распечатки:

INFO: request in request_task: ['hello0', 'world0']
INFO: request in request_task: ['hello1', 'world1']
INFO: request in request_task: ['hello2', 'world2']
INFO: request in request_task: ['hello3', 'world3']
INFO: request in request_task: ['hello4', 'world4']
INFO returning ['hello1', 'world1'] at index 1
INFO: request in request_task: ['hello5', 'world5']
INFO returning ['hello0', 'world0'] at index 0
INFO: request in request_task: ['hello6', 'world6']
INFO returning ['hello6', 'world6'] at index 6
INFO: request in request_task: ['hello7', 'world7']
INFO returning ['hello2', 'world2'] at index 2
INFO: request in request_task: ['hello8', 'world8']
INFO returning ['hello3', 'world3'] at index 3
INFO: request in request_task: ['hello9', 'world9']
INFO returning ['hello4', 'world4'] at index 4
INFO: received sentinel from request_q
INFO returning ['hello9', 'world9'] at index 9
INFO: received sentinel from request_q
INFO returning ['hello5', 'world5'] at index 5
INFO: received sentinel from request_q
INFO returning ['hello7', 'world7'] at index 7
INFO: received sentinel from request_q
INFO returning ['hello8', 'world8'] at index 8
INFO: received sentinel from request_q
INFO: Results have been produced
INFO: data in write_task: ['hello0', 'world0']
INFO: data in write_task: ['hello1', 'world1']
INFO: data in write_task: ['hello2', 'world2']
INFO: data in write_task: ['hello3', 'world3']
INFO: data in write_task: ['hello4', 'world4']
INFO: data in write_task: ['hello5', 'world5']
INFO: data in write_task: ['hello6', 'world6']
INFO: data in write_task: ['hello7', 'world7']
INFO: data in write_task: ['hello8', 'world8']
INFO: data in write_task: ['hello9', 'world9']
INFO: writer has completed!

Обновлять

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

import asyncio
import aiohttp
import aiofiles
import aiocsv
import json

N_REQUEST_TASKS = 5

async def fetch(req: dict) -> dict:
    # Make the request

    # For demo purposes:
    import random

    await asyncio.sleep(random.random())
    result = req['data']
    print(f"INFO returning {result}")
    return result

    async with aiohttp.ClientSession() as session:
        try:
            async with session.request("POST", url=req["url"], data=json.dumps(req["data"]), headers=req["headers"]) as response:
                result = await response.json()
                response.raise_for_status()
                print(f"INFO: response status was: {response.status}")

                # Put response into queue to be written to file
        except Exception as err:
            print(f"ERROR: error making request: {err}")
            result = err
        finally:
            print(f"INFO returning {result}")
            return result


async def create_requests(data_list: list, request_q: asyncio.Queue) -> None:
    # Read tasks from source of data
    for data in data_list:
        # Put a request task into the queue
        req: dict = {
            "headers": {"Accept": "application/json"},
            "url": "https://httpbin.org/post",
            "data": data
        }
        await request_q.put(
            req
        )

    for _ in range(N_REQUEST_TASKS):
        # One sentinel for each request task:
        await request_q.put(None)


async def request_task(request_q: asyncio.Queue, writer_q: asyncio.Queue) -> None:
    while True:
        # Retrieve necessary data to make request
        req = await request_q.get()

        # Sentinel?
        if not req:
            print("INFO: received sentinel from request_q")
            break

        # Make the request which will put data into the response queue
        print(f"INFO: request in request_task: {req['data']}")
        result = await fetch(req)
        await writer_q.put(result)


async def writer(writer_q) -> None:
    async with aiofiles.open("file.csv", mode = "w", newline='') as f:
        w = aiocsv.AsyncWriter(f)

        await w.writerow([
            "status",
            "data",
        ])

        while True:
            result = await writer_q.get()
            if result is None:
                break

            print(f"INFO: data in write_task: {result}")
            if isinstance(result, Exception):
                continue

            # Write the data from the response
            await w.writerow([
                "200",
                json.dumps(result)
            ])
            await f.flush()

async def main() -> None:

    # Create fake data to POST
    data_list: list[str] = [[f"hello{i}", f"world{i}"] for i in range(10)]

    # Request queue
    request_q = asyncio.Queue()
    # Writer queue
    writer_q = asyncio.Queue()

    tasks = []

    # one producer
    tasks.append(
        asyncio.create_task(
            create_requests(data_list, request_q)
        )
    )

    # N_REQUEST_TASKS consumers
    for _ in  range(N_REQUEST_TASKS):
        tasks.append(
            asyncio.create_task(
                request_task(request_q, writer_q)
            )
        )

    writer_task = asyncio.create_task(
        writer(writer_q)
    )

    await asyncio.gather(*tasks)
    print(f"INFO: Results have been produced")

    # Put sentinelto get writer to quit:
    await writer_q.put(None)
    await writer_task
    print("INFO: writer has completed! ")

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