INNER CODE UNIT · Python

archive_batch

lij768423-svg/grok-register-panel · batch_traffic.py:150

def archive_batch(path: str | os.PathLike[str], metrics: object) -> bool:
    entry = _history_entry(metrics)
    if entry is None or entry["bytes_total"] <= 0:
        return False
    target = Path(path)
    ensure_private_dir(target.parent)
    with exclusive_file_lock(target.with_suffix(target.suffix + ".lock")):
        history = read_history(target)
        batches = [
            item
            for item in history["batches"]
            if item.get("batch_id") != entry["batch_id"]
        ]
        batches.append(entry)
        batches.sort(key=_history_sort_key)
        history["batches"] = batches[-HISTORY_LIMIT:]
        atomic_write_json(target, history)
    return True

View source record →