Тысяча INSERT в логах и get_or_create под гонкой: как мы ускорили запись в Django

Вы когда-нибудь смотрели в логи и видели тысячу одинаковых INSERT подряд? Или замечали, что под нагрузкой get_or_create начинает тормозить из‑за лишних round-trip и retry?

В этой статье — про работу с большими объёмами данных в Django ORM. От встроенных bulk_create / bulk_update до PostgreSQL COPY, реальных оптимизаций с прода и самописных атомарных аналогов get_or_create / update_or_create на одном UPSERT.

Я ведущий разработчик в CodeScoring; ниже — то, что мы проходили на реальных сканах зависимостей, а не пересказ документации. Если вы уже знаете, что такое bulk_create, можно начать с подводных камней или с продовых примеров.

Содержание

• Зачем нужны bulk-операции

• Встроенные bulk_create и bulk_update

• Подводные камни bulk

• Работа с конфликтами: ignore_conflicts и update_conflicts

• django-bulk-load и PostgreSQL COPY

• Бенчмарки

• Разность множеств: синхронизация состояния без upsert

• Примеры с прода

• Самописные atomic_get_or_create / update_or_create

• Когда что использовать

• Заключение

Зачем нужны bulk-операции

Представьте кассу в магазине. Вы купили 100 товаров. Кассир может:

1. Пробить каждый товар отдельным чеком — 100 походов к терминалу

2. Пробить всё одним чеком

Второй вариант быстрее не из-за магии, а потому что вы убрали накладные расходы: сетевой round-trip (туда-обратно до БД), разбор SQL, планирование запроса.

То же самое с ORM:

for i in range(1000):
    Product.objects.create(name=f'Product {i}', price=100 + i)

Это тысяча отдельных INSERT. При latency 1 мс вы уже подарили секунду просто на сеть. А ещё индексы, WAL, блокировки. В режиме autocommit каждый create() ещё и отдельная транзакция.

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

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

Встроенные bulk_create и bulk_update

Создание

Наивный подход:

for i in range(1000):
    Product.objects.create(
        name=f'Product {i}',
        price=100 + i,
    )

INSERT INTO products (name, price) VALUES ('Product 0', 100);
INSERT INTO products (name, price) VALUES ('Product 1', 101);
-- ... ещё 998 раз

Bulk:

products = [
    Product(name=f'Product {i}', price=100 + i)
    for i in range(1000)
]
Product.objects.bulk_create(products)

INSERT INTO products (name, price)
VALUES
    ('Product 0', 100),
    ('Product 1', 101),
    -- ...
    ('Product 999', 1099);

Один запрос вместо тысячи.

Обновление

Наивно:

for product in Product.objects.all():
    product.price = new_price(product.price)
    product.save(update_fields=('price',))

Через bulk_update:

products = list(Product.objects.all())
for product in products:
    product.price = new_price(product.price)

Product.objects.bulk_update(products, fields=('price',))

Под капотом Django генерирует что-то вроде:

UPDATE products
SET price = CASE
    WHEN id = 1 THEN 200
    WHEN id = 2 THEN 202
    -- ...
END
WHERE id IN (1, 2, ...);

Работает — до тех пор, пока вы не ждёте от bulk поведения обычного save().

Подводные камни bulk

Сигналы молчат

bulk_create и bulk_update не вызывают сигналы pre_save / post_save и не заходят в Model.save(). Но с auto_now / auto_now_add важный нюанс: это не одно и то же, что сигналы.

При bulk_create Django всё же зовёт Field.pre_save() на полях — и auto_now / auto_now_add проставятся. При bulk_update — нет: в SQL уходят текущие значения атрибутов, поле нужно обновить руками и перечислить в fields.

class Product(models.Model):
    name = models.CharField(max_length=200, unique=True)
    price = models.IntegerField()
    updated_at = models.DateTimeField(auto_now=True)

# bulk_create: updated_at проставится через Field.pre_save
Product.objects.bulk_create(products)

# bulk_update: updated_at сам НЕ обновится
Product.objects.bulk_update(products, fields=('price',))

# Нужно руками
now = timezone.now()
for product in products:
    product.updated_at = now
Product.objects.bulk_update(products, fields=('price', 'updated_at'))

Если на сигнал post_save висит инвалидация кэша или отправка в очередь — bulk это просто проигнорирует. Побочные эффекты придётся делать отдельно или осознанно от них отказаться.

Триггеры на стороне PostgreSQL — другая история: они сработают на реальных INSERT/UPDATE в целевую таблицу. А вот COPY во временную таблицу их обычно не касается. У django-bulk-load нет пути через Field.pre_save, так что auto_now там тоже не «магия» — значения нужно задать на объектах до вызова.

primary key возвращается не всегда

Коротко:

• без конфликтов, на PostgreSQL / MariaDB 10.5+ / SQLite 3.35+, PK после bulk_create обычно появляется в объектах;

• на MySQL / Oracle — чаще нет;

• с ignore_conflicts=True Django PK не проставляет даже на «хороших» БД;

• с update_conflicts=True на PostgreSQL / MariaDB / SQLite PK обычно проставляется (RETURNING не отключается), на MySQL / Oracle — не стоит на это рассчитывать.

db_products = Product.objects.bulk_create(products)
all(p.pk for p in db_products)  # True на PostgreSQL / MariaDB 10.5+ / SQLite 3.35+

db_products = Product.objects.bulk_create(products, ignore_conflicts=True)
any(p.pk for p in db_products)  # False

db_products = Product.objects.bulk_create(
    products,
    update_conflicts=True,
    update_fields=('price',),
    unique_fields=('name',),
)
all(p.pk for p in db_products)  # обычно True на PostgreSQL

Если после вставки с ignore_conflicts нужны id — сделайте повторный SELECT по уникальным ключам.

Другие ограничения Django bulk

Из документации и практики:

• bulk_create не работает с multi-table inheritance и не создаёт M2M;

• bulk_update не обновляет первичный ключ;

• дубликаты PK внутри одного батча bulk_update приводят к тому, что обновится меньше строк, чем вы ждёте;

• даже с batch_size Django сначала готовит SQL/WHEN для батча в памяти — очень большой батч всё равно раздувает клиент.

Размер батча имеет значение

Вставить миллион строк одним INSERT или одним гигантским CASE WHEN — идея спорная. Упираетесь в размер statement, память приложения и длинную транзакцию. Параметр batch_size режет работу на куски: начинайте с 1–5 тысяч, смотрите на время и память, крутите. Без него Django bulk_update на больших объёмах легко превращается в патологический запрос — об этом ниже в бенчмарках.

Работа с конфликтами: ignore_conflicts и update_conflicts

Игнорировать дубликаты

Product.objects.bulk_create(products, ignore_conflicts=True)

INSERT INTO products (name, price)
VALUES ('Product 0', 100), ('Product 1', 101), ...
ON CONFLICT DO NOTHING;

Удобно для идемпотентного импорта: «вставь, что можешь, остальное пропусти».

Нюанс про target: у ON CONFLICT DO NOTHING колонки можно не указывать — тогда тихо проглатывается конфликт по любому unique/exclusion constraint таблицы, не только по name. Именно так Django и генерирует SQL при ignore_conflicts=True на PostgreSQL: голый ON CONFLICT DO NOTHING, без списка колонок. Параметр unique_fields здесь ни при чём — он нужен для update_conflicts. Если unique-индексов несколько, «лишний» конфликт тоже замолчит; чтобы игнорировать только один ключ, пишите raw SQL с явным target или сужайте схему.

Обновлять при конфликте (upsert)

Начиная с Django 4.1:

Product.objects.bulk_create(
    products,
    update_conflicts=True,
    update_fields=('price', 'updated_at'),
    unique_fields=('name',),
)

INSERT INTO products (name, price, updated_at)
VALUES ('Product 0', 100, NOW()), ('Product 1', 101, NOW()), ...
ON CONFLICT (name)
DO UPDATE SET
    price = EXCLUDED.price,
    updated_at = EXCLUDED.updated_at;

Один атомарный запрос: новые строки создаются, существующие обновляются. Для простых справочников — отлично.

Upsert хорош для плоского справочника «вставь или обнови price». Плох — когда нужно ещё закрыть исчезнувшие строки, крутить M2M или не трогать запись, если поля не менялись. Тогда — раздел про разность множеств.

django-bulk-load и PostgreSQL COPY

На больших проектах bulk_create начинает есть время не на INSERT, а на раздувание SQL. Мы пошли в django-bulk-load: она гоняет данные через COPY во временную таблицу, а потом одним-двумя запросами перекладывает в боевую. В коде нас интересовали в основном bulk_insert_models, bulk_update_models и bulk_upsert_models; остальное — в README.

Минусы сразу: только PostgreSQL, завязка на psycopg2, репозиторий маленький — pin версии и не стесняйтесь форкнуть, если застрянете.

Пример вставки:

from django_bulk_load import bulk_insert_models

products = [
    Product(name='Product 1', price=100),
    Product(name='Product 2', price=200),
    Product(name='Product 3', price=150),
]

bulk_insert_models(products)

Схема простая: temp table → COPY → UPDATE/INSERT в целевую → temp исчезает на commit. Если у вас PgBouncer в transaction pooling, с temp tables быстро становится грустно — нужен session pooling или другой путь.

Как на самом деле работает bulk_upsert_models

Слово «upsert» у Django и у django-bulk-load означает разный SQL. Django — один ON CONFLICT DO UPDATE. Cedar — COPY, потом UPDATE ... FROM только где IS DISTINCT FROM, потом INSERT через LEFT JOIN для отсутствующих строк. По умолчанию матч по PK; для справочников по name есть pk_field_names.

Плюс: не затирает created_at и не шлёт гигантский CASE WHEN. Минус: два воркера могут одновременно не увидеть строку и оба полезут в INSERT — при нормальном unique будет IntegrityError, не дубль. Для одного job’а на проект мы живём с этим; для горячего ключа из десяти воркеров — Django ON CONFLICT или advisory lock.

Ограничения

• Только PostgreSQL.

• Сигналы не вызываются. auto_now / auto_now_add сами не проставятся — в отличие от Django bulk_create, здесь нет Field.pre_save.

• Edge case: плохо дружит с ArrayField. Значения для COPY сериализуются как строки, а не через адаптер psycopg. Иногда приходится вручную привести к postgres-литералу массива (ниже — упрощённый пример; для строк с запятыми/кавычками нужна корректная литерализация):

occurrence.files = f'{{{",".join(occurrence.files)}}}' if occurrence.files else None
bulk_insert_models(models=(occurrence,))

На сотнях строк temp table съедает выигрыш — сначала bulk_create с batch_size, COPY потом. Про ArrayField мы узнали по логу malformed array literal — см. хак в примере выше.

Бенчмарки

Цифры ниже — из README django-bulk-load (Cedar), не с нашего прода. Абсолютные секунды зависят от железа и схемы; важен порядок величин.

Важно читать их правильно:

• колонка Django bulk_update — вызов без batch_size, то есть один огромный CASE WHEN. Это патологический сценарий, который Django docs как раз предостерегают не устраивать;

• django-bulk-update — отдельная библиотека (aykut/django-bulk-update), не путать с django-bulk-load;

• bulk_update_models — метод django-bulk-load через COPY.

647 секунд на 100K — это не «Django плохой», а «кто-то вызвал bulk_update без batch_size и получил монстра из CASE WHEN». С нормальным батчингом картина мягче, но на миллионах строк COPY всё равно часто выигрывает — таблицы ниже как раз про порядок величин, не про ваш ноутбук.

Массовое обновление

Массовая вставка

Для больших объёмов на PostgreSQL django-bulk-load стабильно быстрее примерно в 2–3 раза на insert и на порядки — на «ломающем» update без батчинга. На тысяче записей разница копеечная — выбирайте инструмент по масштабу задачи.

Разность множеств: синхронизация состояния без upsert

После скана у вас не «вставить всё подряд», а три корзины: что появилось, что изменилось, что исчезло из актуального снимка (у нас исчезнувшие закрываем через end_date).

Паттерн молча предполагает: на каждый бизнес-ключ не больше одной «текущей» строки (end_date IS NULL). Обычный UniqueConstraint на (project, dependency, license, end_date) этого не даёт: в PostgreSQL два NULL в unique-колонках не равны друг другу, и несколько активных дублей с одним ключом теоретически возможны. Нужен partial unique index — уникальность по полям ключа только при end_date IS NULL:

UniqueConstraint(
    fields=['project', 'dependency', 'license'],
    condition=Q(end_date__isnull=True),
    name='uniq_license_record_current',
)

Тогда set-diff, filter(end_date=None) и «partial unique как предохранитель» из раздела про гонки описывают одну и ту же модель данных.

Наивный код:

for item in incoming:
    obj, created = Model.objects.get_or_create(...)
    if not created:
        obj.save(update_fields=...)

На тысячах объектов это превращается в тысячи SELECT + INSERT/UPDATE. Плюс забытые «закрытия» старых записей.

Паттерн через разность множеств ключей:

existing = {
    (row.dependency_id, row.relation): row
    for row in Model.objects.filter(project=project, end_date=None)
}

actual = {
    (dto.dependency_id, dto.relation): build_model(dto)
    for dto in dtos
}

to_create = actual.keys() - existing.keys()   # новые
to_update = actual.keys() & existing.keys()   # пересечение
to_close = existing.keys() - actual.keys()    # исчезнувшие

Дальше:

# 1. Создаём только новые
Model.objects.bulk_create([actual[k] for k in to_create])

# 2. Обновляем только изменившиеся
changed = []
for key in to_update:
    db_row = existing[key]
    new_row = actual[key]
    if update_if_fields_changed(db_row, new_row, fields):
        changed.append(db_row)
Model.objects.bulk_update(changed, fields=fields)

# 3. Закрываем устаревшие одним UPDATE
Model.objects.filter(
    id__in=[existing[k].id for k in to_close]
).update(end_date=analysis_time)

На десятках тысяч id в IN запрос тяжелеет — тогда лучше temp table / unnest, а не гигантский список в SQL.

Почему upsert здесь часто не замена

• upsert не умеет «закрыть то, чего больше нет во входном наборе»;

• Django ON CONFLICT DO UPDATE перепишет конфликтующие строки даже без реальных изменений; django-bulk-load upsert благодаря IS DISTINCT FROM здесь аккуратнее;

• для истории с end_date IS NULL уникальность часто частичная, и семантика «текущей» записи сложнее простого ON CONFLICT (a, b);

• M2M всё равно требует отдельных insert/delete.

Upsert хорош для плоских справочников. Для синхронизации состояния с create / update / close — разность множеств.

Гонки параллельных сканов

SELECT текущего состояния → diff → bulk-запись не атомарны сами по себе. Два параллельных скана одного проекта могут:

• оба решить «создать» одну и ту же строку;

• один закрыть запись, которую второй ещё считает актуальной;

• затереть чужие обновления.

Что обычно ставят сверху:

• сериализация сканов проекта (очередь / один активный job);

• pg_advisory_xact_lock(project_id) вокруг sync;

• select_for_update() на текущих строках внутри транзакции;

• partial unique index как предохранитель от дублей «текущих» записей.

Без одного из этих слоёв паттерн быстрый, но хрупкий под конкуренцией.

Хелпер «обновить только если реально поменялось»:

def update_if_fields_changed(
    model: models.Model,
    new_data: models.Model,
    update_fields: Iterable[str],
    *,
    allow_update_to_none: bool = False,
) -> bool:
    updated = False
    for field in update_fields:
        existing_value = getattr(model, field, None)
        new_value = getattr(new_data, field)
        can_apply = new_value is not None or allow_update_to_none
        if can_apply and new_value != existing_value:
            setattr(model, field, new_value)
            updated = True
    return updated

Мелочь, а на 100K строк экономит кучу бесполезных UPDATE и не трогает updated_at зря.

Примеры с прода

Сначала мы честно профилировали: этап записи лицензий и зависимостей после SCA тормозил весь скан, в логах — сплошные SELECT/INSERT. Оптимизация select_related тут не спасала: узкое место было в записи, не в чтении. Переписали на set-diff + bulk_insert_models — ниже цифры с типичного проекта на PostgreSQL; у вас будут другие секунды, но порядок ускорения похожий. На LicenseRecord / ProjectDependency в схеме уже стоят partial unique на «текущие» строки (см. выше), без них ускорение bulk не отменяет риск дублей.

Было: get_or_create в цикле

# ~4.84 с
for item in items:
    for license_ in item.licenses:
        record, _ = LicenseRecord.objects.get_or_create(
            project=project,
            dependency=item.dependency,
            license=license_,
            end_date=None,
            defaults={
                'found_at': timezone.now(),
                'start_date': start_date,
            },
        )

Каждая лицензия — отдельный SELECT + возможный INSERT.

Стало: bulk + разность множеств

# ~0.16 с  (примерно x30)
existing = {
    (row.dependency_id, row.license_id): row
    for row in LicenseRecord.objects.filter(project=project, end_date=None)
}

actual = {
    (item.dependency_id, license_.id): LicenseRecord(
        project=project,
        dependency=item.dependency,
        license=license_,
        end_date=None,
        found_at=timezone.now(),
        start_date=analysis_time,
    )
    for item in items
    for license_ in item.licenses
}

to_create = actual.keys() - existing.keys()
if to_create:
    bulk_insert_models(tuple(actual[key] for key in to_create))

# ветка update — как в общем паттерне выше:
# пересечение ключей + update_if_fields_changed + bulk_update
# (для лицензий в этом кейсе поля не менялись — только create/close)

to_close = existing.keys() - actual.keys()
if to_close:
    LicenseRecord.objects.filter(
        id__in=tuple(existing[key].id for key in to_close)
    ).update(end_date=analysis_time)

С ~4.8 с до ~0.16 с на типичном скане проекта в нашем окружении. Тот же результат, другой алгоритм.

Ещё больнее: get_or_create + save + M2M

# ~9.33 с
for item in items:
    dep, created = ProjectDependency.objects.get_or_create(
        project=project,
        dependency=item.dependency,
        relation=item.relation,
        end_date=None,
        defaults={
            'start_date': start_date,
            'vulnerabilities_count': item.vuln_count,
            **extra_fields,
        },
    )
    if not created:
        update_fields = {
            field: value
            for field, value in extra_fields.items()
            if getattr(dep, field) != value
        }
        if update_fields:
            for field, value in update_fields.items():
                setattr(dep, field, value)
            dep.save(update_fields=tuple(update_fields.keys()))

    if item.overridden_licenses:
        dep.overridden_licenses.set(item.overridden_licenses)

Здесь уже не просто N запросов на create, а ещё условные save и .set() на M2M. На больших проектах это легко уезжает в десятки секунд.

Схема «после» та же, что выше, плюс отдельная работа с through-таблицей:

# 1-3. create / update / close для ProjectDependency — как в set-diff выше

# 4. M2M: не .set() в цикле, а bulk по промежуточной таблице
OverriddenLicense.objects.filter(
    project_dependency_id__in=deps_with_new_licenses,
).delete()

bulk_insert_models([
    OverriddenLicense(project_dependency_id=dep_id, license_id=license_id)
    for dep_id, license_ids in new_m2m.items()
    for license_id in license_ids
], ignore_conflicts=True)

Итого снова 3–5 крупных SQL вместо тысяч мелких. В нашем случае время упало примерно на порядок.

Самописные atomic_get_or_create и atomic_update_or_create

Важно: atomic_get_or_create / atomic_update_or_create — не методы Django ORM и не библиотека. Их нет в коробке. Ниже — идея и API, которые мы собрали у себя поверх raw INSERT ... ON CONFLICT ... RETURNING. Хотите так же — пишете хелпер сами (или копируете паттерн). Раздел только про PostgreSQL; на MySQL / SQLite xmax и этот трюк не заведутся. Если параллельных воркеров на один unique нет — обычного get_or_create достаточно, можно пролистать.

Чтобы не путать с ORM: у Django get_or_create не трогает существующую строку через defaults; update_or_create — трогает (с 5.0 ещё create_defaults только на insert). Наши функции повторяют эту семантику снаружи, но внутри — один UPSERT, а не SELECT + INSERT + retry.

В чём реальная боль get_or_create

Документация Django честна: метод атомарный, если на поиске стоит unique constraint. При гонке двух INSERT один получит IntegrityError, Django поймает его внутри и сделает повторный get(). При наличии unique constraint наружу ошибка обычно не вылетает.

Django не «врёт» — просто miss-путь это SELECT + INSERT, под гонкой ещё retry по IntegrityError, а в APM это выглядит как мусор. Дубли появляются, когда unique есть только в голове, а в миграции — нет.

update_or_create идёт через select_for_update().get_or_create(): существующую строку он обновляет аккуратнее, но на пути «строки ещё нет» INSERT-гонка никуда не делась. Плюс лишние блокировки. С Django 5.0 у метода также есть create_defaults — значения только для создания, отдельно от defaults для обновления.

Идея: один UPSERT + RETURNING

PostgreSQL умеет атомарно:

INSERT INTO dependency (name, language, version, purl, homepage)
VALUES ('requests', 'python', '2.32.0', 'pkg:pypi/requests@2.32.0', 'https://...')
ON CONFLICT (name, language, version)
DO UPDATE SET id = dependency.id  -- no-op, чтобы сработал RETURNING
RETURNING *, (xmax = 0) AS created;

Так за один round-trip получаем и строку, и флаг created.

Свой atomic_get_or_create

Это не Model.objects.…, а обычная функция у вас в проекте. Семантика близка к get_or_create, но одним UPSERT: собрать колонки для INSERT, подобрать conflict target (включая partial unique), выполнить INSERT ... ON CONFLICT DO UPDATE SET pk = pk RETURNING *, (xmax = 0) AS created, собрать модель из строки. Полная реализация длинная из‑за partial index и разбора RETURNING — в статье оставляем контракт и SQL; код пишете под свою схему.

Как это выглядит снаружи, когда хелпер уже есть:

dependency, created = atomic_get_or_create(
    Dependency,
    name=package.name,
    language=package.language,
    version=package.version,
    purl=package.purl,
    defaults={
        'homepage': 'https://example.com',
    },
)

Conflict target здесь — (name, language, version), а не все kwargs подряд: purl уходит в INSERT, но не в условие конфликта. Это сознательное отличие от Django get_or_create, где любой kwarg кроме defaults — часть lookup. Если строка уже есть — вернётся она, defaults не перезапишут существующие поля (как в классическом get_or_create).

Свой atomic_update_or_create

Та же самописная обёртка, но в ветке конфликта реально обновляются поля из defaults. Если defaults пустой — отдельная ветка без UPDATE, как get_or_create.

dependency, created = atomic_update_or_create(
    Dependency,
    name=package.name,
    language=package.language,
    version=package.version,
    purl=package.purl,
    defaults={
        'homepage': info.homepage,
        'description': info.description,
    },
)

SQL уже с реальным обновлением:

INSERT INTO dependency (...)
VALUES (...)
ON CONFLICT (name, language, version)
DO UPDATE SET
    homepage = EXCLUDED.homepage,
    description = EXCLUDED.description
RETURNING *, (xmax = 0) AS created;

Если значения те же, этот UPDATE всё равно создаст новую версию строки (см. выше про dead tuples). Чтобы не плодить их зря, можно сузить апдейт через WHERE ... IS DISTINCT FROM ... — но тогда при «ничего не изменилось» Postgres не вернёт строку в RETURNING. Нужен fallback SELECT, CTE, или no-op DO UPDATE без WHERE.

Если defaults пустой — поведение как у get_or_create (без UPDATE существующих полей), но через отдельную ветку кода, а не через no-op upsert.

Глубже: partial unique index и NULL

Можно пропустить, если у вас нет nullable-поля в unique и нет гонок на таком ключе.

В PostgreSQL по умолчанию в unique-ограничении NULL не равен NULL: две строки с (dependency_id=1, repository_id=NULL) обе пройдут обычный UNIQUE (dependency_id, repository_id).

Это поведение сохранится, пока вы не используете UNIQUE NULLS NOT DISTINCT — он появился в PostgreSQL 15. На более старых версиях обычный unique с nullable-полем дырявый.

Классическое решение — два частичных индекса:

class PackageScan(models.Model):
    dependency = models.ForeignKey(Dependency, on_delete=models.CASCADE)
    repository = models.ForeignKey(Repository, null=True, on_delete=models.CASCADE)

    class Meta:
        constraints = [
            UniqueConstraint(
                name='uniq_repo_not_null',
                fields=['dependency', 'repository'],
                condition=Q(repository__isnull=False),
            ),
            UniqueConstraint(
                name='uniq_repo_is_null',
                fields=['dependency'],
                condition=Q(repository__isnull=True),
            ),
        ]

Для ON CONFLICT нужно указать не только колонки, но и WHERE индекса:

-- repository задан
ON CONFLICT (dependency_id, repository_id)
WHERE repository_id IS NOT NULL
DO UPDATE SET id = package_scan.id

-- repository IS NULL
ON CONFLICT (dependency_id)
WHERE repository_id IS NULL
DO UPDATE SET id = package_scan.id

Самописные хелперы умеют выбрать нужный partial constraint по значениям вставляемой строки. Без этого ON CONFLICT просто не попадёт в индекс — и вы снова ловите гонки.

Альтернатива на PostgreSQL 15+: один constraint с NULLS NOT DISTINCT и более простой ON CONFLICT. Но миграция существующих данных и совместимость со старыми инстансами — отдельная история; partial unique indexes остаются рабочим и переносимым вариантом.

Почему не select_for_update

select_for_update защищает обновление существующей строки. Он не создаёт строку и не спасает от двух параллельных INSERT. UPSERT решает обе ветки одной командой, без лишних блокировок «на всякий случай».

На практике под конкурентной записью одного и того же ключа (20 потоков одновременно) такие UPSERT-хелперы стабильно создают ровно одну строку и ровно один раз возвращают created=True. Это стоит покрывать TransactionTestCase с барьером потоков — именно тот тест, который ловит гонки, а не «зелёный на одном воркере».

Когда что использовать

Не надо везде тащить самописный UPSERT. Но там, где два воркера бьются за один ключ, обычный ORM API дорогой и шумный — даже когда формально корректен.

Заключение

Не усложняйте заранее. Сотня строк в цикле с save() — нормально. Тысячи однотипных записей — bulk, на PostgreSQL и жирных объёмах имеет смысл COPY. Нужно синхронизировать «снимок» с create/update/close — разность множеств, не слепой upsert. Несколько воркеров бьются за один ключ — настоящий ON CONFLICT, а не надежда, что retry спасёт.

Перед merge я обычно смотрю три вещи: не повесили ли на bulk ожидание сигналов (и не забыли ли про auto_now на bulk_update / COPY); нужны ли PK сразу после insert и какие флаги конфликтов; и есть ли lock или сериализация, если sync не один поток.

Django ON CONFLICT и upsert в django-bulk-load называются одинаково, но внутри разные звери. Стоит увидеть конкретный SQL и цену round-trip — и становится ясно, какой инструмент брать. Без магии и без сюрпризов.

Только зарегистрированные пользователи могут участвовать в опросе. Войдите, пожалуйста.