Spaces:
Sleeping
Sleeping
File size: 3,375 Bytes
5c14256 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | """
insert_guard.py — Atomic insert guard for all challenge generators.
Guarantees that dedup check + DB insert are serialised per (table, team_role),
eliminating the race condition where two concurrent inserts both pass the
dedup check and create a duplicate.
Usage:
from app.services.insert_guard import atomic_insert
async def refill_pool(...):
...
ok = await atomic_insert("encryption_challenges", team_role, row,
is_duplicate_func, [title, story])
"""
from __future__ import annotations
import asyncio
from collections import defaultdict
from typing import Any, Callable, Optional
_TABLE_LOCKS: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
def _lock_key(table: str, team_role: str) -> str:
return f"{table}::{team_role}"
async def atomic_insert(
table: str,
team_role: str,
row: dict,
dedup_func: Optional[Callable[..., bool]] = None,
dedup_args: Optional[list] = None,
dedup_kwargs: Optional[dict] = None,
) -> bool:
"""Insert *row* into *table* after a serialised dedup check.
Parameters
----------
table : str
Supabase table name (e.g. ``"encryption_challenges"``).
team_role : str
``"red"`` or ``"blue"`` — used to scope the lock.
row : dict
The full DB row to insert.
dedup_func : callable, optional
A synchronous function that returns ``True`` if a duplicate exists.
Called **under the lock** so the check and insert are atomic.
dedup_args : list, optional
Positional args for *dedup_func*.
dedup_kwargs : dict, optional
Keyword args for *dedup_func*.
Returns
-------
bool
``True`` if the row was inserted, ``False`` if skipped (duplicate)
or on error.
"""
key = _lock_key(table, team_role)
lock = _TABLE_LOCKS[key]
# Re-use the same httpx client across calls within the lock scope
import httpx
from app.services.supabase_service import supabase_headers
from app.core.config import SUPABASE_URL, SUPABASE_ANON_KEY
if not SUPABASE_ANON_KEY or not SUPABASE_URL:
return False
async with lock:
# 1. Dedup check (under lock)
if dedup_func is not None:
try:
is_dup = dedup_func(
*(dedup_args or []),
**(dedup_kwargs or {}),
)
if is_dup:
title = row.get("title", "?")[:60]
print(f" [insert_guard] Dedup skipped: {title} in {table}")
return False
except Exception as e:
print(f" [insert_guard] Dedup check error (non-fatal): {e}")
# 2. Insert (under lock)
url = f"{SUPABASE_URL}/rest/v1/{table}"
headers = supabase_headers(content_type=True)
headers["Prefer"] = "return=representation"
try:
async with httpx.AsyncClient(timeout=20) as client:
resp = await client.post(url, json=row, headers=headers)
if resp.status_code in (200, 201):
return True
print(f" [insert_guard] DB insert {resp.status_code}: {resp.text[:200]}")
return False
except Exception as e:
print(f" [insert_guard] DB insert error: {e}")
return False
|