43 lines
1.8 KiB
Python
43 lines
1.8 KiB
Python
|
"""Shared helper for check-then-act get-or-create races (section 15/28).
|
||
|
|
|
||
|
|
Every get_or_create_* across the repository layer follows check-then-act
|
||
|
|
(look up by natural key, insert if missing) rather than a single atomic
|
||
|
|
UPSERT. That's fine as long as nothing ingests concurrently for
|
||
|
|
overlapping data — which isn't a safe assumption once more than one
|
||
|
|
ingestion job can run at the same time (arq's own concurrency, season
|
||
|
|
ingestion cascading via `include_seasons`, and this bulk-import script's
|
||
|
|
worker pool all create exactly this situation): two concurrent callers can
|
||
|
|
both see "missing" and both try to INSERT, and the loser's INSERT hits a
|
||
|
|
unique-constraint violation instead of finding the winner's row.
|
||
|
|
|
||
|
|
`create_or_recover` wraps the insert in a SAVEPOINT and, on that specific
|
||
|
|
failure, rolls back just the savepoint and re-queries for what the
|
||
|
|
concurrent caller committed — recovering instead of crashing. Originally
|
||
|
|
lived only in `reference_repository.py`; moved here once
|
||
|
|
`provider_repository.py:get_or_create_provider` needed the exact same
|
||
|
|
protection (found live: 50 concurrent bulk-import workers all racing to
|
||
|
|
create the one-and-only "tmdb" Provider row on first run crashed 49 of
|
||
|
|
them with `uq_providers_slug` violations)."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from collections.abc import Awaitable, Callable
|
||
|
|
|
||
|
|
from sqlalchemy.exc import IntegrityError
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
|
||
|
|
|
||
|
|
async def create_or_recover[T](
|
||
|
|
session: AsyncSession,
|
||
|
|
create: Callable[[], Awaitable[T]],
|
||
|
|
recover: Callable[[], Awaitable[T | None]],
|
||
|
|
) -> T:
|
||
|
|
try:
|
||
|
|
async with session.begin_nested():
|
||
|
|
return await create()
|
||
|
|
except IntegrityError:
|
||
|
|
existing = await recover()
|
||
|
|
if existing is None:
|
||
|
|
raise
|
||
|
|
return existing
|