从列表中迭代创建 URL 列表,并为请求调用每个 URL

Iteratively create a list of URLS from a list and call each for a request

我正在尝试迭代创建 urls 组合两个 url 变量和需要添加到 url 的唯一 uuid 列表,就像这样 .com/{test_adv_uuid}/.我当前的代码如下所示:

import logging
import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

order_id = 1
order_amount = 10

test_adv_uuid = [
    f'08d79951-1908-4f0f-91aa-40eee5ac0e3f',
    f'30816356-4be2-4def-9793-2913dc7dae82',
    f'40a9420a-3bad-4778-930d-0e3da35355d1',
    f'9c2c4477-21ea-4b90-b72b-db40c8ae9754',
    f'6f0fa70b-4914-458d-8b02-0b6a4475773f',
    f'9614bd9f-afa0-4d93-b709-beb38c99cd66',
]


test_adv_url_body = f'https://listen.whatever.com/{test_adv_uuid}/pixel.png'
test_adv_url_attr = f'https://listen.whatever.com/{test_adv_uuid}/pixel.png?order={order_id}&value={order_amount}'


# look up thread pool stuff for understanding on the download & logging 
THREAD_POOL = 16

session = requests.Session()
session.mount(
    'https://',
    requests.adapters.HTTPAdapter(pool_maxsize=THREAD_POOL,
                                  max_retries=3,
                                  pool_block=True)
)

def get(url):
    response = session.get(url)
    logging.info("request was completed in %s seconds [%s]", response.elapsed.total_seconds(), response.url)
    if response.status_code != 200:
        logging.error("request failed, error code %s [%s]", response.status_code, response.url)
    if 500 <= response.status_code < 600:
        # server is overloaded? give it a break
        time.sleep(5)
    return response

def download(urls):
    with ThreadPoolExecutor(max_workers=THREAD_POOL) as executor:
        # wrap in a list() to wait for all requests to complete
        for response in list(executor.map(get, urls)):
            if response.status_code == 200:
                print(response.content)

def main():
    logging.basicConfig(
        format='%(asctime)s.%(msecs)03d %(levelname)-8s %(message)s',
        level=logging.INFO,
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    

    urls = [
        test_adv_url_body,
        test_adv_url_attr
    ]

    download(urls)

if __name__ == "__main__":
    main()

我得到的输出将列表中的所有 uuid 项目组合成一个 url,看起来像:

request was completed in 0.286232 seconds [https://listen.whatever.com/%5B'08d79951-1908-4f0f-91aa-40eee5ac0e3f',%20'30816356-4be2-4def-9793-2913dc7dae82',%20'40a9420a-3bad-4778-930d-0e3da35355d1',%20'9c2c4477-21ea-4b90-b72b-db40c8ae9754',%20'6f0fa70b-4914-458d-8b02-0b6a4475773f',%20'9614bd9f-afa0-4d93-b709-beb38c99cd66'%5D/pixel.png?order=1&value=10]

我将如何重构它以迭代地使这些 url 中的每一个与动态 uuid 成为一个单独的请求?

我会在列表中迭代创建 url 吗?还是在请求调用本身内?

试试这个

# better style to name your lists with plurals
test_adv_uuids = [
    '08d79951-1908-4f0f-91aa-40eee5ac0e3f',
    '30816356-4be2-4def-9793-2913dc7dae82',
    '40a9420a-3bad-4778-930d-0e3da35355d1',
    '9c2c4477-21ea-4b90-b72b-db40c8ae9754',
    '6f0fa70b-4914-458d-8b02-0b6a4475773f',
    '9614bd9f-afa0-4d93-b709-beb38c99cd66',
]
# use a list comprehension to build the list of urls
test_adv_url_bodies = [f'https://listen.whatever.com/{uuid}/pixel.png' for uuid in test_adv_uuids]
test_adv_url_attrs = [f'https://listen.whatever.com/{test_adv_uuid}/pixel.png?order={order_id}&value={order_amount}' for uuid in test_adv_uuids]

然后在你的 main() 中你会

urls = test_adv_url_bodies + test_adv_url_attrs

这在技术上比在单个 for 循环中创建 urls 列表效率略低,而不是使用两个列表推导,所以我将把它留作 reader