Как сэкономить время, деньги и нервы на простых задачах

В данной статье я расскажу, как можно решить интеграционную задачу с помощью простых инструментов: Python, Lambda и пары бутылок пива за три часа.

В качестве предисловия.

Разработка новых сервисов и продуктов в энтерпрайзе, как правило, требует достаточно сложного производственного цикла: описание задачи, анализ со стороны архитектора, системных аналитиков, оценка, постановка в бэклог, реализация, тестирование и проч. проч.проч. В зависимости от сложности задачи с момента ее возникновения до момента раскатки на прод могут пройти месяцы.

Я так не люблю. Годы работы в бизнес-подразделениях (кредитование, офферинг и ML) научили работать быстро в ограниченных ресурсах и искать альтернативные пути решения (денег нет, но вы держитесь).

Поэтому, когда ко мне прилетела факультативная задачка с оценкой от IT в 3 месяца, о которой я расскажу ниже, я малость удивился: а что там делать-то эти три месяца?

Но начнем с самого начала.

Прилетела ко мне задачка на подумать: необходимо сделать так, чтобы курсы наших обменников, которые опубликованы на сайте, появились на известном банковском сервисе и периодически обновлялись. Это было даже не ТЗ, а ФТ. Окей, думаю я, давайте по порядку. Какой источник данных внутри, какая структура источника, какой протокол, где API, куда стучаться? Бизнес-заказчик, естественно, на технические вопросы ответить не может. Ему нужен сервис. И у бизнес-заказчика есть только пример формата итоговых данных от «той стороны», который разрабатываемый сервис должен обеспечить: XML-файл с определенной структурой. При чем, без wsdl, что поначалу немного смутило.

Ладно. XML-файлы мы уже умеем гонять и в REST и SOAP. Давайте копать и изучать Jira, как-никак, задача там заводилась.

0. Nothing. Null Komma nichts.

В том смысле, что задача не заводилась в Jira. Собираемся, берем инициативу в свои руки и начинаем этой инициативой долбать заказчика. В итоге выясняется следующее:

Источник данных по курсам находится в одной из закрытых БД, где необходимо спроектировать вьюху, которая будет отдавать нужные данные, и получить к ней доступ под технической учеткой. Т.е., нужно поставить задачу ДБА.
У нас есть кабинет на целевом сервисе, в котором есть дополнительные примеры и документация по сервисам. Отлично.
Приемник на сервисе не использует ни REST ни SOAP, а самостоятельно читает файлы по предоставленной ссылке с какой-то периодичностью. Т.е., нужен эндпоинт, который будет отдавать файл по прямой ссылке. Это усложняет задачу.
Дополнительно нужно засёрвить отдельную HTML-страницу с курсами центрального допофиса. Т.е., нужен веб-сервер. Задачка стала тяжелее на порядок.

ШТОШ, давайте есть слона бутербродами.

Первое. На чем реализовывать. Я использую для разработки Python по ряду причин. Главные — наличие 100500 уже реализованных проверенных библиотек под любую задачу, а также простота и читаемость кода (если придерживаться рекомендаций PEP8).

Дальше архитектура. Если подходить в лоб, то нужно внутри домена найти машину, которая по расписанию будет крутить скрипт. Скрипт должен подключаться к БД, читать данные, преобразовывать их и формировать файлы. Плюс там же нужно поднять простенький HTTP-сервер, который будет сёрвить одну страницу и один файл. Многовато для такого скрипта.

Если чуть-чуть ужать аппетиты, то можно обойтись докером на сварме и какой-нибудь хранилкой. Ключевое слово — «какой-нибудь». Нет их. В том смысле, что можно сделать NFS или SMB-шару, но вместе с такой шарой открыть потенциальную дырку во внутреннюю сеть. На такое мне пойти родина и совесть не позволяют.

А теперь давайте посмотрим с другой стороны.Курсы публикуются на сайте и обновляются несколько раз в день. Т.е., это публичная информация и, по идее, я могу не завязываться на внутренний источник и брать инфо прямо с сайта. Прекрасно. Beautiful.… BeautifulSoup. Есть такая библиотечка на питоне. Умеет читать и валидировать HTML из кода. Значит, код можно вынести из домена и никому ничего за это не будет. Плюсов у такого решения масса:

данные в сервисе синхронизированы с публичной информацией и не зависят от состояния внутренней БД с курсами. Даже если что-то случается с БД и данные на сайте отображаются из кэша — мы публично никого не обманываем.
устраняем необходимость в зависимости от очередного тяжеловеса в лице SQLite и драйвера к БД.
Убираем звено в лице ДБА из цепочки разработки и поддержки.

Осталось решить вопрос с запуском самого скрипта и эндпоинтами.

Сам по себе скрипт не хранит и не использует информацию о состоянии, а это прямая дорога в Lambda. А где Lambda, там и S3 для хранения.

Ну вот и вырисовывается архитектура: лямбда, которая читает по триггеру курсы с сайта, лопатит их в XML и сохраняет в S3 с веб-сервером. Осталось выбрать, где это все запускать.

С AWS у меня отношения сложились достаточно давно, я в нем реализовывал несколько экспериментальных сервисов, которые потом раскатывал on-prem. Но тут коллега предложил попробовать Yandex, т.к. наша корпоративная архитектура двигается в сторону облачных технологий с большим ускорением (более подробно можно почитать тут). Плюс ко всему, характеристики скрипта и эндпоинта укладываются в бесплатный уровень: скрипт запускается по крону не чаще одного раза в пять минут, потребляет не более 128 мегабайт памяти (это минимум, который можно указать для скрипта) и сохраняет где-то 100 килобайт данных. Помимо всего прочего в бакетах яндекса можно включить хостинг.

Ну яндекс так яндекс. Это не реклама, просто так получилось.

Сам скрипт удалось написать за вечер. Из внешних зависимостей понадобились:

BeautifulSoup4 для парсинга страницы с курсами. Сама по себе библиотека не сложная для освоения. По сути получаем объект «страница » и через методы find и find_all ищем по заданным параметрам на странице div с таблицами и внутри него получаем список таблиц (и вспомогательных атрибутов):

page = BeautifulSoup(session.get(url, headers=headers).text.encode('utf-8'), features="lxml") full_table = page.find('div', {'id': 'cash-rates-accordion'}) all_dos = full_table.find_all('div', {'class': 'highlighted'}) all_tabs = full_table.find_all('table')

Requests. Как таковой requests изначально не планировался в явном виде. BS использует его как зависимость и импортирует самостоятельно. Но в моем случае хотелось в явном виде отправлять кастомные заголовки для дебага и установить количество повторов с настраиваемым временем перезапуска при ошибках доступа:

session = requests.Session() retry = Retry(total=9, backoff_factor=0.15) adapter = HTTPAdapter(max_retries=retry) session.mount('https://', adapter)

Pandas для работы с таблицами. Можно было бы уменьшить потребляемую память за счет отказа от pandas, но тогда количество кода увеличилось бы в разы, т.к. исходные данные представлены в табличном виде:

for dop, tab in zip(all_dos, all_tabs): dftab = pd.read_html(str(tab), header=0)[0]

Ну и для обратного преобразования таблицы в HTML (подзадачка с формированием отдельной страницы для центрального обменника):

str_io = io.StringIO() table.to_html(buf=str_io, index=False, border=False) html_table = str_io.getvalue()

Boto3. Yandex Cloud.Storage предоставляет S3-совместимые бакеты. Грех этим не воспользоваться:

session = boto3.session.Session() s3 = session.client( service_name='s3', endpoint_url='https://storage.yandexcloud.net', region_name=os.environ['AWS_REGION'], aws_access_key_id=os.environ['AWS_ACCESS_KEY_ID'], aws_secret_access_key=os.environ['AWS_SECRET_ACCESS_KEY'] ) def writer(bucket, file, filetype, buffer): response = s3.put_object(Bucket=bucket, Key=file, Body=buffer.getvalue()) status = response.get("ResponseMetadata", {}).get("HTTPStatusCode") if status == 200: return f"Successful S3 put_object response on {filetype}. Status - {status}" else: return f"Unsuccessful S3 put_object response on {filetype}. Status - {status}" with io.BytesIO() as buffer: tree.write(buffer, encoding='utf-8', xml_declaration=True, method='xml') print(writer(os.environ['BUCKET_NAME'], filename, filetype, buffer))

Да, при миграции из AWS в Yandex много кода переписывать не придется))

Из стандартной библиотеки пригодились:

os для чтения переменных окружения (не прописывать же ключи доступа в коде):

region_name=os.environ['AWS_REGION'], aws_access_key_id=os.environ['AWS_ACCESS_KEY_ID'], aws_secret_access_key=os.environ['AWS_SECRET_ACCESS_KEY']

re для задачи установления разных курсов для разных объемов валюты:

def get_unit(name): try: result = int(re.search(r"от (\d+)", name).group(1)) # выводим коэффициент в виде найденного объема валюты except AttributeError: result = 1 # выводим коэффициент 1, если в строке нет объема return result

io для создания разного рода и вида буферов. Если уж использовать лямбду, то без записи промежуточных состояний на диск:

with io.BytesIO() as buffer: buffer.write(html.encode())

xml для решения главной задачи. Внешний сервис потребляет xml, и для генерации простых xml ничего кроме стандартной библиотеки не нужно:

def build_xml(tab, mapping_curs): df = tab.copy(deep=True) df['cur_code'] = df['Валюта'].map(mapping_curs) df['quantity'] = df['Название валюты'].apply(lambda x: get_unit(x)) office = Element('office') office.set('code', tab.name) office.set('show_rates', "1") currencies = SubElement(office, 'currencies') for name, group in df.groupby(by='cur_code'): currency = Element('currency') currency.set('id', str(name)) exchanges = SubElement(currency, 'exchanges') for quant, subgroup in group.groupby(by='quantity'): exchange = Element('exchange') amount = SubElement(exchange, 'amount') amount.text = str(quant) buy = SubElement(exchange, 'buy') buy.text = str(round(subgroup.iloc[0]['Курс покупки'] / subgroup.iloc[0]['Количество единиц'], 4)) sale = SubElement(exchange, 'sale') sale.text = str(round(subgroup.iloc[0]['Курс продажи'] / subgroup.iloc[0]['Количество единиц'], 4)) exchanges.append(exchange) currencies.append(currency) return office t_group = Element('group') t_group.set('version', "1.0") for table in tables: t_group.append(build_xml(table, mapping_curs)) tree = ET.ElementTree(t_group) ET.indent(tree, '\t')

По коду приблизительно так. Для создания лямбды на яндексе можно загрузить архив с. py и requirements.txt, можно написать их прямо в онлайн-редакторе. Можно подзаморочиться и написать terraform для раскатки (что я в итоге и сделал, ибо CI/CD), правда, он у яндекса еще далек от совершенства, в т.ч. по документации.

Что касается остальной инфраструктуры, тут покажу на примере кусков терраформа:

Готовим код функции локально. Не забываем рядом положить requirements.txt:

data "archive_file" "source" { type = "zip" source_dir = local.root_dir output_path = "/tmp/function-${local.timestamp}.zip" }

Создаем саму функцию из архива и пропихиваем ей в системные переменные ключи от сервисной учетной записи, из-под которой она будет ходить в storage:

resource "yandex_function" "rates-function" { name = local.function_name description = "Prepares XML and HTML files using site parsing" entrypoint = local.function_entry_point memory = "128" runtime = "python39" execution_timeout = "60" service_account_id = yandex_iam_service_account.app-sa.id content { zip_filename = data.archive_file.source.output_path } environment = { AWS_REGION = local.region AWS_ACCESS_KEY_ID = yandex_iam_service_account_static_access_key.sa-static-key.access_key AWS_SECRET_ACCESS_KEY = yandex_iam_service_account_static_access_key.sa-static-key.secret_key BUCKET_NAME = local.target_bucket } user_hash = data.archive_file.source.output_md5 }

Триггер. Обычный крон:

resource "yandex_function_trigger" "every-5-min" { name = "rates-function-5-min" description = "Fire rates function every 5 minutes" timer { cron_expression = "0/5 * ? * * *" } function { id = yandex_function.rates-function.id service_account_id = yandex_iam_service_account.app-sa.id } }

Бакет. Заодно настраиваем ACL и включаем веб-сервер. В local.target_bucket задаем уникальное имя, или терраформ выдаст ошибку, если найдет в яндексе такое имя (и если предварительно не дропнуть этот ресурс):

resource "yandex_storage_bucket" "rates-bucket" { access_key = yandex_iam_service_account_static_access_key.sa-static-key.access_key secret_key = yandex_iam_service_account_static_access_key.sa-static-key.secret_key bucket = local.target_bucket acl = "public-read" website { index_document = "rates.html" } }

Блок с сервисным аккаунтом:

resource "yandex_iam_service_account" "app-sa" { name = "rates-sa" } resource "yandex_function_iam_binding" "sa-invoker" { members = ["serviceAccount:${yandex_iam_service_account.app-sa.id}"] role = "serverless.functions.invoker" function_id = yandex_function.rates-function.id } resource "yandex_resourcemanager_folder_iam_member" "app-sa-editor" { folder_id = local.folder_id role = "storage.editor" member = "serviceAccount:${yandex_iam_service_account.app-sa.id}" } resource "yandex_resourcemanager_folder_iam_binding" "app-sa-uploader" { folder_id = local.folder_id members = ["serviceAccount:${yandex_iam_service_account.app-sa.id}"] role = "storage.uploader" } resource "yandex_resourcemanager_folder_iam_binding" "app-sa-invoker" { folder_id = local.folder_id members = ["serviceAccount:${yandex_iam_service_account.app-sa.id}"] role = "serverless.functions.invoker" } resource "yandex_iam_service_account_static_access_key" "sa-static-key" { service_account_id = yandex_iam_service_account.app-sa.id description = "static access key for object storage" }

Для внешнего сервиса необходимо было прибивать гвоздями имена эндпоинтов, поэтому перед раскаткой терраформом пришлось дополнительно запускать terraform destroy для бакета. Остальные объекты могут создаваться с timestamp или хэшем в имени.

Ну и куда без тестирования. Питон привлекателен тем, что содержит удобные фреймворки для написания автотестов.

from hypothesis import given, settings, strategies as st, Verbosity import re def get_unit(name): try: result = int(re.search(r"от (\d+)", name).group(1)) except AttributeError: result = 1 return result currencies = ['Швейцарский франк', 'Японская йена', 'Евро', 'Фунт стерлингов', 'Доллар США', 'Шведская крона', 'Норвежская крона', 'Китайский юань'] @settings(max_examples=100, verbosity=Verbosity.verbose) @given(x=st.sampled_from(currencies), y=st.integers(min_value=0, max_value=10000)) def test_get_unit(x, y): if y in (0, 1): assert get_unit(x) == 1 else: string = ''.join([x, ' (от ', str(y), ')']) assert get_unit(string) == y

Один из тестов для функции, которая используется в основном коде. Тут я применил библиотеку hypothesis, которая позволяет сгенерировать необходимые наборы тестовых данных по заданным стратегиям.

Подробно останавливаться на тестировании не буду, это отдельная тема для разговора.

На самом деле, работа была закончена еще задолго до терраформа. Функция раскатана в частном облаке, триггеры настроены, эндпоинты настроены, ссылки прописаны. Сервис заработал на следующий день после начала работы. Были мелкие допилы в части оптимизации кода и справочников, но в целом задача была решена практически за один вечер вместо трех месяцев, и заказчик остался доволен.

Осталось решить парочку административных вопросов с суппортом и критериями критичности сервиса, и можно полностью закрывать задачу.