spark_on_k8s.cli.receipts

Implementation reference. These symbols describe how the project works; their presence here does not make them supported application imports. Use the documented entry points and support policy.

receipt_writer(destination, *, field='run')

Check local storage before submission and publish without replacing an existing file.

Source code in spark_on_k8s/cli/receipts.py

@contextmanager
def receipt_writer(destination: Path | None, *, field: str = "run"):
    """Check local storage before submission and publish without replacing an existing file."""
    if field not in {"run", "server"}:
        raise ValueError("Unsupported receipt field")
    if destination is None:
        yield lambda run: None
        return
    destination = destination.expanduser().absolute()
    if os.path.lexists(destination):
        raise FileExistsError("Receipt destination already exists; choose a new file")
    descriptor, temporary = tempfile.mkstemp(prefix=".spark-receipt-", dir=destination.parent)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:

            def write(run):
                json.dump({"schema_version": 1, field: run.to_dict()}, stream)
                stream.write("\n")
                stream.flush()
                os.fsync(stream.fileno())
                # Hard-link publication is atomic and never replaces an existing path.
                os.link(temporary, destination)

            yield write
    finally:
        os.unlink(temporary)