python异步函数正确的语法

Question

我正在尝试更正我的异步功能，以便使用以下脚本：

import asyncio
import platform
import aiohttp

if platform.system() == 'Windows':
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

url = 'https://services.rappi.com.br/api/ms/web-proxy/dynamic-list/cpgs/'

req_headers = {'authority': 'services.rappi.com.br', 'accept': 'application/json',
               'authorization': 'Bearer ft.gAAAAABjMbGKkc2fkTMa2M2EKuBrCM1Z1vU5Ww1Fw03CjpJEb9UF1DO1TAjwpAD0H0NIImuMjWFcOkUURseLzJIi0DNSOr-oRWcZWgcnHLm2Ed6rDLvxQ2ikdGLtyVXZRqgGHWOMlBPVSKjYLb6NMZmAeHhAsGNjiQ3vP5VEdb_ULA9S5Lpo8H7-ElhKufmlVqQ6CrDyTUsyQeZ3IzbNCbN8MBLFhgRxVMZSwyl640YXF9ZvQUI1sibP-Ko86xrin_2EXEmAdEk7aSl9u0ezlmnBL6Wk8a7CwSJUEUAwAjrNdJTLSodjQaiVVx7TZ0rQEkzPgceaH7wtpmvl--6txmRDnBu4g0na3Km19K1LNzs0fz7-_Go8Qlg=',
               'deviceid': '957933bd-cdc4-4876-ad00-bc00a1d55c29',
               'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36'}

json_data = {"dynamic_list_request":
    {
        "limit": 100, "offset": 0, "context": "sub_aisles",
        "state": {"lat": "-23.5516221", "lng": "-46.7404627"}
    },
    "dynamic_list_endpoint": "context/content",
    "proxy_input":
        {
            "seo_friendly_url": "900597707-assai-atacadista-super-sao-paulo",
            "aisle_friendly_url": "",
        }
}

json_data2 = {"dynamic_list_request":
    {
        "limit": 100, "offset": 0, "context": "aisle_detail",
        "state": {"lat": "-23.5516221", "lng": "-46.7404627"}
    },
    "dynamic_list_endpoint": "context/content",
    "proxy_input":
        {
            "seo_friendly_url": "900597707-assai-atacadista-super-sao-paulo",
            "aisle_friendly_url": "bebidas",  
            'subaisle_friendly_url': "" 
        }
}
market_list = ['900597707-assai-atacadista-super-sao-paulo', '900520986-makro-atacadista-sao-paulo']
data = [{'sub_category': 'temperos-e-molhos', 'category': 'mercearia'},
        {'sub_category': 'massas', 'category': 'mercearia'}]


async def fetch(url, session):
    list_items = []
    global data
    for cats in data:  # 1) >> how to pass here to ASYNC? <<
        json_data2['proxy_input']['subaisle_friendly_url'] = cats['sub_category']
        json_data2['proxy_input']['aisle_friendly_url'] = cats['category']
        for markets in market_list:  # 2) >> how to pass here to ASYNC? <<
            json_data2['proxy_input']['seo_friendly_url'] = markets
            print((json_data2['proxy_input']['subaisle_friendly_url'], json_data2['proxy_input']['aisle_friendly_url'],
                   json_data2['proxy_input']['seo_friendly_url']))
            async with session.post(url, headers=req_headers, json=json_data2) as response:
                try:
                    json_body = await response.json()
                    json_body = json_body['dynamic_list_response']['data']['components']
                except KeyError:
                    json_body = None
                if json_body is not None:
                    for sections in json_body:
                        for items in sections['resource']['products']:
                            list_items.append(items)
    return list_items


async def main():
    tasks = []
    async with aiohttp.ClientSession() as session:
        tasks.append(
            asyncio.create_task(fetch(url, session))
        )
        body = await asyncio.gather(*tasks)
    return body


print(asyncio.run(main()))

我知道我的session.post上的for循环不起作用，因为它们不是异步的，那么如何纠正这些循环呢？我已经试过很多东西了，但都没用。

async with session.post(url, headers=req_headers, json=json_data2) as response:

Obs:代码正在工作，如果需要，可以运行并测试它。

• 整个代码可以更改，如果需要，我不介意.

Answer 1

为了实现异步并行，应该将内部for循环的主体重构到另一个可以作为索引任务运行的协同例程中。通过这种方式，异步主循环可以并行地编排各种调用：

下面的代码可能会变得更加优雅，方法是继续向任务提供“猫”，并让一些任务使用asyncio.wait而不是gather并行运行。“轻松”重构就是将对单个类别的所有调用放在一个任务列表中，并收集以下内容：

未测

async def fetch_items(session, markets):
    local_items = []
    json_data2['proxy_input']['seo_friendly_url'] = markets
    print((json_data2['proxy_input']['subaisle_friendly_url'], json_data2['proxy_input']['aisle_friendly_url'],
            json_data2['proxy_input']['seo_friendly_url']))
    async with session.post(url, headers=req_headers, json=json_data2) as response:
        try:
            json_body = await response.json()
            json_body = json_body['dynamic_list_response']['data']['components']
        except KeyError:
            return
        for sections in json_body:
            local_items.extend(items for items in sections['resource']['products'])
    return local_items
    

async def fetch(url, session):
    list_items = []
    global data
    for cats in data:  
        json_data2['proxy_input']['subaisle_friendly_url'] = cats['sub_category']
        json_data2['proxy_input']['aisle_friendly_url'] = cats['category']
        pending_tasks = [asyncio.create_task(fetch_items(session, markets)) for markets in market_list]
        for result in await asyncio.gather(*pending_tasks):
            list_items.extend(result)

    return list_items


async def main():
    tasks = []
    async with aiohttp.ClientSession() as session:  # there is no repetition here: no point in using .gather
        body = await fetch(url. session) 
    return body