Итак, в своем последнем посте я рассказал о том, как асинхронность может ускорить выполнение задач и сократить время их выполнения. Сегодня мы немного испачкаем руки.
Мы будем использовать один из моих любимых API — DiceBear Avatar API. Он принимает исходную строку (на самом деле любой случайный текст) и возвращает изображение профиля со случайными функциями. Есть несколько стилей на выбор, и мы можем точно настроить, какие функции мы хотели бы видеть.
Пример использования с использованием curl:

curl -X GET https://avatars.dicebear.com/api/big-smile/:seed.png --output DiceBearAvatar.png

Мы собираемся создать программу на Python, которая автоматизирует этот процесс и асинхронно сгенерирует столько изображений профиля, сколько нам нужно. Чтобы иметь немного контекста, мы собираемся создать синхронную версию программы.

Для начала нам нужно установить пару библиотек:

pip install aiohttp requests

Нам понадобятся вспомогательные функции, которые мы поместим в utils.py. Они будут обрабатывать такие вещи, как запись данных изображения в файл, форматирование URL-адресов для вызовов API и создание случайных строк для нашей программы.

# ./utils.py
# --------------------------------
import secrets
from os import path

RANDOM_SEED_SIZE = 10

  
def generate_random_seed():
 return secrets.token_urlsafe(RANDOM_SEED_SIZE)

def generate_api_url(seed, avatar_style, base_url, extension=".png"):
 url = base_url.format(avatar_style=avatar_style, seed=seed+extension)
 return url

  
  
def write_to_file(data: str, file_name: str, file_path: str = "my_profile_pics"):
 try:
  with open(path.join(file_path, file_name+".png"), "wb") as file:
   file.write(data)

 except Exception as e:
  print(f"\n{e}\n")

Далее будут фактические HTTP-запросы, и сейчас мы будем использовать requests library.

# ./sync_api.py
# --------------------------------
import requests
import utils  


def create_new_avatar(avatar_style:str, base_url:str):
 seed = utils.generate_random_seed()
 request_url = utils.generate_api_url(seed=seed, avatar_style=avatar_style, base_url=base_url)
 resp = requests.get(request_url)
 data = resp.content
 utils.write_to_file(data=data, file_name=seed, file_path=r"C:/Users/user/desktop/pp")

Просто стандартные HTTP-запросы. Мы используем наши служебные функции, чтобы сгенерировать случайное начальное число для DiceBear и сделать запрос GET. Затем мы извлекаем данные изображения из результирующего объекта ответа.
Запись этих данных в реальный файл выполняется другой утилитой. Если вы хотите изменить место хранения файла, вы можете отключить C:/Users/user/desktop/pp на все, что вы предпочитаете.

Выглядит уже хорошо, и мы почти закончили.

Теперь, чтобы создать точку входа для программы, main.py.

# ./main.py
# ---------------------------
import sync_api
from timeit import default_timer as timer

BASE_URL = "https://avatars.dicebear.com/api/{avatar_style}/{seed}"

ALL_STYLES = [

 "adventurer",

 "adventurer-neutral",

 "avataaars",

 "big-ears",

 "big-ears-neutral",

 "big-smile",

 "botts",

 "croodles",

 "micah"

 ]

def generate_profile_pics(n):
 for _ in range(n):
  sync_api.create_new_avatar(avatar_style="adventurer", base_url=BASE_URL)


start = timer()
generate_profile_pics(20)
print(f"Time Elapsed: {timer()-start} seconds")

Вот и все! Мы настроили его для создания 20 изображений профиля для нас.

Примечание. DiceBearAvatar API — отличный бесплатный ресурс, которым не следует злоупотреблять. Используйте ответственно!

Вы можете пойти дальше и запустить:

python main.py

Вы должны получить 20 изображений PNG в указанном вами каталоге.

Хорошо, теперь к хорошему. Есть ли способ ускорить этот кусок кода? Можем ли мы запустить несколько экземпляров этой задачи одновременно? Введите асинхронные библиотеки asyncio и aiohttp, наш набор инструментов для выполнения асинхронных веб-запросов в Python.

aiohttp лучше всего работает с клиентским сеансом для обработки нескольких запросов, поэтому мы будем использовать его (requests также поддерживает клиентские сеансы, но это не популярная парадигма).

# ./async_api.py
# ----------------------------
import aiohttp
import asyncio
import utils

async  def aio_create_new_avatar(client_session:aiohttp.ClientSession, avatar_style:str, base_url:str):
 seed = utils.generate_random_seed()
 request_url = utils.generate_api_url(seed=seed, avatar_style=avatar_style, base_url=base_url)

 async with client_session.get(request_url) as resp:
  data = await resp.read()
 utils.write_to_file(data=data, file_name=seed, file_path=r"C:/Users/user/desktop/pp")

Хорошо. Так что не слишком отличается от sync_api.py. У нас есть еще пара импортов, а затем мы используем диспетчер контекста с сеансом клиента, чтобы сделать веб-запрос. Если синтаксис async/await для вас в новинку, вы можете прочитать этот пост, в котором представлена ​​вся идея асинхронности в Python.

Далее мы собираемся изменить main.py, чтобы использовать наш новый код. Мне нравится хорошая гонка, поэтому мы собираемся отслеживать время выполнения как асинхронного, так и синхронного кода.
Вот обновленный main.py:

# ./main.py
# -----------------------------------
# new imports
import aiohttp
import asyncio
import async_api

import sync_api

from timeit import default_timer as timer

BASE_URL = "https://avatars.dicebear.com/api/{avatar_style}/{seed}"

ALL_STYLES = [

 "adventurer",

 "adventurer-neutral",

 "avataaars",

 "big-ears",

 "big-ears-neutral",

 "big-smile",

 "botts",

 "croodles",

 "micah"

 ]

def generate_profile_pics(n):

 for _ in range(n):
  sync_api.create_new_avatar(avatar_style="adventurer", base_url=BASE_URL)

  
async def async_generate_profile_pics(n):

 Client = aiohttp.ClientSession()
 Tasks = []

 for _ in range(n):

  # Create a new profile pic
  Tasks.append(
    async_api.aio_create_new_avatar(
       client_session=Client, 
       avatar_style="big-smile",
       base_url=BASE_URL
    )
  )

  try:
   await asyncio.gather(*Tasks)  

  except:
   pass

  finally:
   await Client.close()

  
start = timer()
generate_profile_pics(20)
print(f"Time Elapsed: {timer()-start} seconds")

# For Windows Users 
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

start = timer()
asyncio.run(async_generate_profile_pics(20))
print(f"Time Elapsed: {timer()-start} seconds")

Сделанный! И если мы запустим это, мы должны получить еще 40 изображений профиля в указанном нами каталоге.
В среднем синхронный код выполнялся на моем компьютере около 16 секунд, а асинхронный — около секунды! Основные улучшения времени, если вы спросите меня. Так что в данном случае компромисс между сложностью кода и его оптимизацией окупился.

Этот проект был особенно полезен для меня. Я использую его для создания отображаемых фотографий для личного использования, а также в качестве замещающих отображаемых фотографий для веб-приложений, с которыми я работаю.
Если вы хотите использовать эти изображения в коммерческих целях, ознакомьтесь с лицензиями для каждого стиля. .

Полный код этого проекта можно найти здесь.









Спасибо за прочтение! Прощальный мем 🙃