Хочу рассказать о своей разработке и поделиться ей с сообществом habr в виде готового модуля для Python.

Этот модуль решает задачу обновления данных в базе данных. Он выполняет insert/update/delete в одном действии.

Модуль DBMerge проектировался для упрощения ETL процессов по загрузке данных из множества внешних источников в SQL базу.

Для взаимодействия с базой данных используется SQLAlchemy и ее универсальные механизмы, таким образом, разработка не имеет привязки к конкретной БД. (На момент написания статьи, детальные тесты проводились для PostgreSQL, MariaDB, SQLite, MS SQL.)

Принцип действия

Принцип действия следующий: модуль создает временную таблицу в базе и записывает в нее весь входящий датасет через операцию массового insert. Далее он выполняет инструкции UPDATE, INSERT, DELETE для целевой таблицы.

схема работы модуля dbmerge
схема работы модуля dbmerge

Но на самом деле, конечно, все не так просто, поэтому модуль имеет много параметров для поддержки разных сценариев.

Сначала поговорим об исходных данных. Поддерживаются 3 варианта источников данных:

Из pandas

Этот вариант реализован для случая когда мы забираем данные в DataFrame (например, из csv), там их как-то преобразуем и очищаем, а потом закидываем в БД.

Из list of dict

Это вариант удобен, когда мы не хотим использовать pandas или когда нужно вставить данные, которые неудобно обрабатывать в DataFrame, например UUID или JSONB (dict).

Из таблицы или представления

Этот вариант нужен в случае, если у нас есть "тяжелое" представление (view) и мы хотим периодически сохранять результат его выполнения в целевой таблице и п�� ней уже строить запросы. В принципе, это похоже на materialized view в postgres, но тут мы можем обновлять данные частями.

Установка 

pip install dbmerge

Как пользоваться

with dbmerge(engine=engine, data=data, table_name="Facts",
             delete_mode='delete', <доп параметры>) as merge:
   merge.exec(<доп параметры>)

Создаем объект в блоке with, указав engine, ваши данные (data), целевую таблицу (table_name) и другие параметы.

На странице github есть примеры кода и детальное описание всех параметров.

Далее вызываем exec() для выполнения обновления данных. В exec() можно передать параметры отбора для удаления/пометки удаления записей и параметры выборки из исходной таблицы, если она используется.

Теперь хочу рассказать об алгоритме работы

Проверка метаданных

Модуль проверяет, существует ли целевая таблица, какой в ней ключ, есть ли в ней поля, содержащиеся в исходных данных.

Если есть новые поля, то пытается определить их тип из data_types в параметрах или из самих данных. 

Если таблицы нет или полей не хватает, то модуль создаст таблицу или досоздаст поля в имеющейся таблице.

Технические детали:

  • В параметрах можно задать схемы для целевой таблицы (schema), для временной таблицы (temp_schema) и для исходной таблицы (source_schema). В SQLite понятие схемы не поддерживается, поэтому для данной БД настройки будут проигнорированы. В MariaDB понятие схемы отличается, но модуль, в любом случае, со схемами будет работать.

  • Если каких-то полей нет в таблице, то для их создания желательно задать тип данных, например, String(250). Для MariaDB это придется делать для строковых полей, т.к. эта БД требует указывать длину поля. Для других БД модуль сам может определить данные как String() и поле будет создано без задания типа.

Создание временной таблицы

Имя временной таблицы будет таким же как и у целевой таблицы, но к нему будет прибавлен случайный хэш. При выходе из конструкции with или при завершении выполнения exec(), эта таблица будет автоматически удалена.

Технические детали:

  • Ключ таблицы (key) можно задать в параметрах при создании объекта dbmerge. Если таблица уже существует, то делать этого не нужно, т.к. модуль сам определит, какой у нее первичный ключ. Если в таблице нет первичного ключа, то можно все же сообщить его в параметре key, но для нормального быстродействия обновления данных, конечно, первичный ключ или индекс нужны. Совсем без ключа модуль работать не сможет и, при его отсутствии, поднимет exception.

  • Временная таблица также будет создана с первичным ключом, т.к. бывают ситуации, когда БД "хочет" его использовать, а также он нужен для проверки входящих данных. Понятно, что ключевые поля во входящих данных, не могут содержать повторы или значения None или np.nan.

  • Для MS SQL операция массого insert во временную таблицу происходит значительно медленнее, чем в другие БД. Проблема в библиотеке pyodbc, но думаю есть надежда, что сделают поддержку SQLAlchemy для новой библиотеки mssql-python.

Обновление изменившися записей

Тут мы делаем в запросе соединение временной таблицы с целевой таблицей, и сравниваем значения в неключевых полях. Если есть отличия, то делаем update.

Если задано поле merged_on_field, то записываем туда текущую дату и время.

Технические детали:

  • Модуль не обновляет данные, если они не отличаются от данных целевой таблицы. Это существенно снизит нагрузку на базу, если ваш сценарий предполагает частую загрузку данных, в которых меняется небольшой процент записей. 

  • Дата и время, указанные в merged_on_field, будут показывать тот момент, когда были обнаружены и записаны отличия в данных, независимо от того, сколько раз вы делали загрузку.

  • Для записи даты и времени используется функция now() в SQLAlchemy, которая вызывает NOW() в БД. Часовой пояс будут зависеть от параметров вашего подключения. Само поле в БД может быть типа timestamp как с часовым поясом так и без.

  • Для MariaDB, при сравнении строковых значений, по умолчанию нет отличий между строчными и прописными буквами, также не играют роли начальные/конечные пробелы. Например, 'Test ' = 'test' будет Истина и обновления данных не произойдет. MS SQL не различает строчные и прописные буквы, но ловит отличие в начальных/конечных пробелах. Если вам нужна тут точность, тогда нужно сделать изменение настроек collation в БД (Для PostgreSQL и SQLite такой проблемы нет).

  • Сравнение отличий на самом деле производится через функцию IS DISTINCT FROM для того, чтобы корректно обрабатывать значения NULL.

Вставка отсутствующ��х записей

Эта часть выполняется как insert из select записей, которых нет в целевой таблице.

Если в параметрах задано поле inserted_on_field, то в это поле запишется текущая дата и время. Если задано merged_on_field, то оно также запишется при вставке.

Удаление или пометка удаления "пропавших" записей

Предусмотрено несколько режимов удаления (delete_mode):

- no - по умолчанию, ничего не делаем.

- delete - удаляем из целевой таблицы данные остутствующие в исходной таблице

- mark - устанавливаем флаг удаления (delete_mark_field) для данных, отсутствующих в исходной таблице.

Для вариантов delete или mark можно задать критерии отбора (delete_condition).

Например, вы обновляете данные периодами и вам нужно, чтобы были удалены "исчезнувшение" строки в загружаемом периоде, но при этом мы не трогаем другие периоды. Тогда вызов merge будет выглядеть, например, так:

with dbmerge(engine=engine, data=data, table_name="Facts",
            delete_mode='delete', merged_on_field='Merged On') as merge:
   merge.exec(delete_condition=
              merge.table.c['Date'].between(date(2025,2,1),date(2025,2,28)))

Параметр delete_condition мы задаем уже после того, как инициализировали объект merge и в нем создались метаданные. В атрибуте table будет содержаться объект Table из SQLAlchemy, в котором мы находим, например, столбец Date и ставим по нему критерий between.

Фактически здесь мы пишем логический критерий отбора в формате SQLAlchemy, который будет использован в where для команды delete. Я привел несколько чуть более сложных примеров таких отборов в коде на github.

Замер быстродействия

Приведу результаты небольшого замера приозводительности но ноутбуке с Ubuntu 24.04. (БД и Python на одной машине).

время merge по разным БД
время merge по разным БД

Соотношение операций при тесте: insert - 65%, update - 7%, delete - 15%, не изменилось - 13%

Плохой результат MS SQL связан, думаю, с библиотекой pyodbc, которая не умеет делать массовый insert.

Заключение

Я думаю, что получившийся модуль лучше всего подойдет для процессов ETL и интеграции данных.

Для малого количества строк, возможно, схема с временной таблицей является избыточной, но зато она позволяет не трогать целевую таблицу, если ничего не изменилось.

Я с интересом жду мнений профессионального сообщества в комментариях к данной статье.