INNER CODE UNIT · Python

finalize_batch

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

def finalize_batch(path: str | os.PathLike[str], batch_id: str, exit_code: int) -> dict:
    close_runtime(path=path, batch_id=batch_id)
    metrics = read_metrics(path)
    if metrics.get("batch_id") != str(batch_id):
        return metrics
    now = time.time()
    metrics.update(
        {
            "running": False,
            "updated_at": now,
            "finished_at": now,
            "active_connections": 0,
            "exit_code": int(exit_code),
        }
    )
    atomic_write_json(path, metrics)
    return metrics

View source record →