"""Shared DB-access helpers for the async services. `get_or_create` centralizes the race-safe find-or-create dance — SELECT, then on a miss a savepoint INSERT that recovers (NOT a full rollback) when a concurrent worker inserted the same row first. It was hand-rolled identically in ArtistService, TagService and ExtensionService; divergent copies are exactly how the duplicate-row / race bugs in [[reference_scalar_one_or_none_duplicates]] crept in, so it lives in one place now (DRY pattern sweep 2026-06-09). Note: this is the ASYNC sibling of `Importer._get_or_create` (sync, used by the filesystem-import path). The two can't share an implementation across the sync/async boundary; the importer one stays as the lone sync consumer. """ from __future__ import annotations from collections.abc import Awaitable, Callable from sqlalchemy import Select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession async def get_or_create[T]( session: AsyncSession, select_stmt: Select, factory: Callable[[], Awaitable[T]], ) -> tuple[T, bool]: """Race-safe find-or-create. Returns ``(row, created)``. Run ``select_stmt`` (scalar_one_or_none); if a row exists, return it with ``created=False``. Otherwise open a SAVEPOINT and ``await factory()`` — which must add its row(s), flush, and return the primary row. On ``IntegrityError`` (a concurrent worker inserted the same row first) roll back the SAVEPOINT — NOT the outer transaction, which would lose the caller's surrounding work — and re-run ``select_stmt`` (scalar_one) to return the row the other worker created. The caller owns the outer commit. A UNIQUE/partial-unique constraint matching ``select_stmt``'s predicate is required for the recovery to trip; without it a duplicate slips through. """ existing = (await session.execute(select_stmt)).scalar_one_or_none() if existing is not None: return existing, False sp = await session.begin_nested() try: row = await factory() await sp.commit() return row, True except IntegrityError: await sp.rollback() return (await session.execute(select_stmt)).scalar_one(), False