Function _write_rollup

api/usage_log.py:261–291

(batch: list[tuple])

async def _write_rollup(batch: list[tuple]) -> None:
    counts: dict[tuple[str, str, str, str], int] = defaultdict(int)
    for ts, user_id, service, _host in batch:
        counts[(user_id, ts.strftime("%Y-%m"), service, ts.strftime("%d"))] += 1

    # Group the batch by hash first, so chunking splits between hashes rather
    # than through the middle of one and every EXPIRE lands with its HINCRBYs.
    by_key: dict[str, list] = defaultdict(list)
    for (user_id, month, service, day), n in counts.items():
        by_key[_rollup_key(user_id, month)] += [_rollup_field(service, day), n]

    global _rollup_script
    redis = await get_redis()
    if _rollup_script is None:
        _rollup_script = redis.register_script(_ROLLUP_LUA)

    items = list(by_key.items())
    for start in range(0, len(items), ROLLUP_CHUNK):
        chunk = items[start:start + ROLLUP_CHUNK]
        # Keys are addressed by index from ARGV so each hash is named once,
        # however many of its fields the batch touches.
        keys = [k for k, _ in chunk]
        argv: list = [ROLLUP_TTL_DAYS * 86400]
        for i, (_key, fields) in enumerate(chunk, start=1):
            for j in range(0, len(fields), 2):
                argv += [i, fields[j], fields[j + 1]]
        # Each chunk is independent. A failure part-way through leaves earlier
        # chunks applied, which is the same partial-batch outcome the caller
        # already tolerates - these counters are explicitly approximate, and
        # HINCRBY is not idempotent so retrying would double-count.
        await _rollup_script(keys=keys, args=argv)


async def _flush(batch: list[tuple]) -> None: