Python · API · Просунутий рівень

Асинхронні запити з httpx

5 завдань

Використовуйте httpx з asyncio для паралельних запитів до API та прискорення скриптів із кількома викликами.

Короткий вступ до теми та пояснення перед вправами (вправи нижче):

httpx та asyncio

#
Бібліотека `requests` виконує запити послідовно -- кожен наступний починається тільки після завершення попереднього: ``` GET /products/1/ → чекаємо → GET /products/2/ → чекаємо → ... ``` Якщо кожен запит займає 200 мс, то 10 запитів -- це 2 секунди. Асинхронні запити виконуються паралельно: ``` GET /products/1/ ┐ GET /products/2/ ├→ всі виконуються одночасно → чекаємо всіх GET /products/3/ ┘ ``` Ті самі 10 запитів завершуються за ті ж ~200 мс. `httpx` -- це бібліотека, схожа на `requests`, з підтримкою асинхронного режиму. У синхронному режимі вона працює так само, як `requests`: ```python import httpx # Синхронно -- те саме що requests response = httpx.get('https://apilearn.tukas.dev/api/products/') print(response.json()['count']) ``` Для асинхронного режиму використовуйте `AsyncClient` з `async/await`: ```python import httpx import asyncio async def fetch_products(): async with httpx.AsyncClient() as client: response = await client.get('https://apilearn.tukas.dev/api/products/') return response.json() asyncio.run(fetch_products()) ``` `async with` відкриває клієнт і автоматично закриває його по завершенні блоку. `await` зупиняє поточну корутину, поки не прийде відповідь, але не блокує цикл подій -- тому інші запити можуть виконуватися паралельно. `asyncio.gather()` запускає кілька корутин одночасно і чекає на всі: ```python results = await asyncio.gather( client.get(url1), client.get(url2), client.get(url3), ) # results -- список об'єктів Response в тому самому порядку ``` Коли використовувати async: - Багато незалежних запитів (всі категорії, всі сторінки, деталі продуктів). - Запити не залежать від результатів один одного. - Кожен запит має помітну затримку (мережеві виклики майже завжди мають). Async не допомагає, коли запити послідовні за природою (наприклад, потрібен результат запиту 1, щоб знати що питати у запиті 2).

Практичні приклади asyncio.gather

#
Практичні приклади паралельних запитів із httpx: Отримати кількість продуктів у всіх категоріях паралельно: ```python import httpx import asyncio BASE_URL = 'https://apilearn.tukas.dev' async def fetch_category_count(client, slug): response = await client.get( f'{BASE_URL}/api/products/', params={'category': slug, 'page_size': 1}, ) return slug, response.json()['count'] async def main(): async with httpx.AsyncClient() as client: r = await client.get(f'{BASE_URL}/api/categories/') categories = r.json()['results'] slugs = [cat['slug'] for cat in categories] tasks = [fetch_category_count(client, slug) for slug in slugs] results = await asyncio.gather(*tasks) for slug, count in results: print(f'{slug}: {count}') asyncio.run(main()) ``` Отримати кілька продуктів за id паралельно: ```python import httpx import asyncio BASE_URL = 'https://apilearn.tukas.dev' async def fetch_product(client, product_id): response = await client.get(f'{BASE_URL}/api/products/{product_id}/') return response.json() async def main(): async with httpx.AsyncClient() as client: products = await asyncio.gather(*[ fetch_product(client, i) for i in range(1, 6) ]) for p in products: print(p['name'], p['price']) asyncio.run(main()) ``` `*[...]` розгортає список у позиційні аргументи для `asyncio.gather`. `gather(*tasks)` еквівалентний `gather(task1, task2, task3, ...)`. Примітка щодо httpx vs requests: `httpx` підтримує і синхронний, і асинхронний режими, має майже ідентичний API до `requests` і підтримує HTTP/2. Для суто синхронних скриптів `requests` цілком підходить. Використовуйте `httpx`, коли потрібен async.

Довідник httpx та asyncio

#
Порівняння httpx і requests: ``` requests httpx Синхронний режим requests.get() httpx.get() Асинхронний режим — httpx.AsyncClient Сесія requests.Session httpx.Client (синх) / httpx.AsyncClient API запитів .get .post .json() однакове raise_for_status ✓ ✓ ``` Паралельні запити з asyncio.gather: ```python # Список корутин → gather → список результатів (у тому ж порядку) responses = await asyncio.gather( client.get(url1), client.get(url2), ) # responses[0] відповідає url1, responses[1] — url2 ``` Розгортка списку з `*`: ```python # Замість: await asyncio.gather(fetch(1), fetch(2), fetch(3), fetch(4), fetch(5)) # Використовуйте: await asyncio.gather(*[fetch(i) for i in range(1, 6)]) ``` Коли що використовувати: ``` requests Прості скрипти, послідовні запити, невелика кількість запитів httpx async Багато незалежних запитів, швидкість важлива ```

Вправи:

01

Перший асинхронний запит

#

Напишіть асинхронну функцію з httpx.AsyncClient, яка отримує /api/products/ і виводить кількість продуктів. Запустіть її через asyncio.run().

import httpx
import asyncio

BASE_URL = 'https://apilearn.tukas.dev'

async def fetch_products():
    # Використайте httpx.AsyncClient щоб отримати /api/products/
    pass

asyncio.run(fetch_products())
Рішення 
import httpx
import asyncio

BASE_URL = 'https://apilearn.tukas.dev'

async def fetch_products():
    async with httpx.AsyncClient() as client:
        response = await client.get(f'{BASE_URL}/api/products/')
        data = response.json()
        print(data['count'])

asyncio.run(fetch_products())
02

Паралельні запити з gather

#

Використайте asyncio.gather() щоб зробити два запити паралельно: GET /api/products/ і GET /api/categories/. Виведіть загальну кількість продуктів і кількість категорій.

import httpx
import asyncio

BASE_URL = 'https://apilearn.tukas.dev'

async def main():
    async with httpx.AsyncClient() as client:
        # Зробіть обидва запити паралельно через gather
        pass

asyncio.run(main())
Рішення 
import httpx
import asyncio

BASE_URL = 'https://apilearn.tukas.dev'

async def main():
    async with httpx.AsyncClient() as client:
        products_resp, categories_resp = await asyncio.gather(
            client.get(f'{BASE_URL}/api/products/'),
            client.get(f'{BASE_URL}/api/categories/'),
        )
    print(products_resp.json()['count'])
    print(len(categories_resp.json()))

asyncio.run(main())
03

Отримати кілька продуктів паралельно

#

Напишіть корутину fetch_product(client, product_id), яка отримує /api/products/{id}/. Використайте asyncio.gather() щоб отримати продукти з id від 1 до 5 одночасно. Виведіть назву і ціну кожного.

import httpx
import asyncio

BASE_URL = 'https://apilearn.tukas.dev'

async def fetch_product(client, product_id):
    # GET /api/products/{product_id}/
    pass

async def main():
    async with httpx.AsyncClient() as client:
        # Зберіть і виконайте 5 запитів через gather
        pass

asyncio.run(main())
Рішення 
import httpx
import asyncio

BASE_URL = 'https://apilearn.tukas.dev'

async def fetch_product(client, product_id):
    response = await client.get(f'{BASE_URL}/api/products/{product_id}/')
    return response.json()

async def main():
    async with httpx.AsyncClient() as client:
        products = await asyncio.gather(*[
            fetch_product(client, i) for i in range(1, 6)
        ])
    for p in products:
        print(p['name'], p['price'])

asyncio.run(main())
04

Авторизований асинхронний клієнт

#

Використайте httpx.AsyncClient із заголовком авторизації (отриманим через синхронний requests.post). Зробіть паралельні GET до /api/cart/ і /api/users/profile/. Виведіть username і кількість товарів у кошику.

import httpx
import asyncio
import requests

BASE_URL = 'https://apilearn.tukas.dev'

# 1. Отримайте токен через requests (синхронно)
# 2. Використайте AsyncClient з заголовком Authorization
# 3. Паралельно отримайте профіль і кошик
Рішення 
import httpx
import asyncio
import requests

BASE_URL = 'https://apilearn.tukas.dev'

token = requests.post(f'{BASE_URL}/api/auth/token/', json={
    'username': 'ВАШ_ЛОГІН', 'password': 'ВАШ_ПАРОЛЬ',
}).json()['token']

async def main():
    headers = {'Authorization': f'Token {token}'}
    async with httpx.AsyncClient(headers=headers) as client:
        profile_resp, cart_resp = await asyncio.gather(
            client.get(f'{BASE_URL}/api/users/profile/'),
            client.get(f'{BASE_URL}/api/cart/'),
        )
    print(profile_resp.json()['username'])
    print(len(cart_resp.json()['items']))

asyncio.run(main())
05

Порівняння послідовного та паралельного виконання

#

Виміряйте час виконання двох підходів для отримання продуктів 1-5: спочатку з requests у циклі (послідовно), потім з httpx і asyncio.gather (паралельно). Виведіть час для обох варіантів.

import httpx
import asyncio
import requests
import time

BASE_URL = 'https://apilearn.tukas.dev'

# Підхід 1: послідовний з requests
start = time.time()
# ... ваш код ...
print(f'Послідовний: {time.time() - start:.2f}s')

# Підхід 2: паралельний з httpx
start = time.time()
# ... ваш код ...
print(f'Паралельний: {time.time() - start:.2f}s')
Рішення 
import httpx
import asyncio
import requests
import time

BASE_URL = 'https://apilearn.tukas.dev'

# Підхід 1: послідовний
start = time.time()
for i in range(1, 6):
    requests.get(f'{BASE_URL}/api/products/{i}/')
print(f'Послідовний: {time.time() - start:.2f}s')

# Підхід 2: паралельний
async def fetch(client, i):
    return await client.get(f'{BASE_URL}/api/products/{i}/')

async def parallel():
    async with httpx.AsyncClient() as client:
        await asyncio.gather(*[fetch(client, i) for i in range(1, 6)])

start = time.time()
asyncio.run(parallel())
print(f'Паралельний: {time.time() - start:.2f}s')