Function _write_archive

api/usage_log.py:241–258

(batch: list[tuple])

async def _write_archive(batch: list[tuple]) -> None:
    global _seq
    _seq += 1
    # Keyed by flush time, not by the events' own timestamps: a batch spans up
    # to BATCH_INTERVAL and can straddle midnight, so an object dated today may
    # hold up to a minute of yesterday. Every row carries its own `ts`, so a
    # reader filtering on that is exact; a reader filtering on the key prefix
    # should widen the range by a day at each end.
    now = datetime.now(timezone.utc)
    body = await asyncio.to_thread(
        lambda: gzip.compress(_encode_batch(batch), compresslevel=6)
    )
    key = r2_store.usage_key(
        now.strftime("%Y-%m-%d"), HOST_ID, now.strftime("%H%M%S"), RUN_TOKEN, _seq
    )
    await r2_store.put_bytes(
        key, body, content_type="application/x-ndjson", content_encoding="gzip"
    )


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