fileio

fileio

Atomic file I/O helpers shared across the data pipelines.

The frontend and backend processes share one cache directory (a Docker bind mount in production) and the frontend's file watcher reloads pipeline outputs whenever their mtime changes. Plain writes leave a window where the file is truncated or half-written, so a concurrent reader can crash on a torn parquet or a crash mid-write can corrupt the file permanently. These helpers always write to a temp file in the destination's directory and then move it into place with os.replace (atomic within one filesystem).

write_parquet_atomic(df, path)

Write a DataFrame as parquet without exposing a torn file to readers.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame to persist.	required
path	Path	Destination parquet path.	required

Source code in pspcz_analyzer/fileio.py

def write_parquet_atomic(df: pl.DataFrame, path: Path) -> None:
    """Write a DataFrame as parquet without exposing a torn file to readers.

    Args:
        df: DataFrame to persist.
        path: Destination parquet path.
    """
    _atomic_write(path, lambda tmp: _write_parquet(df, tmp))

write_json_atomic(data, path)

Write a JSON-serializable object atomically (indented, trailing newline).

Parameters:

Name	Type	Description	Default
data	Any	JSON-serializable payload.	required
path	Path	Destination JSON path.	required

Source code in pspcz_analyzer/fileio.py

def write_json_atomic(data: Any, path: Path) -> None:
    """Write a JSON-serializable object atomically (indented, trailing newline).

    Args:
        data: JSON-serializable payload.
        path: Destination JSON path.
    """
    _atomic_write(path, lambda tmp: _write_json(data, tmp))

stream_to_atomic(response, dest, chunk_size=_CHUNK_SIZE)

Stream an open httpx response body to dest atomically.

A failure mid-stream (connection reset, timeout) removes the temp file and leaves dest absent rather than truncated, so the next run downloads cleanly instead of tripping over a partial file.

Parameters:

Name	Type	Description	Default
response	Response	Response from client.stream(...), used inside its context manager.	required
dest	Path	Destination file path.	required
chunk_size	int	Read chunk size in bytes.	_CHUNK_SIZE

Source code in pspcz_analyzer/fileio.py

def stream_to_atomic(response: httpx.Response, dest: Path, chunk_size: int = _CHUNK_SIZE) -> None:
    """Stream an open httpx response body to ``dest`` atomically.

    A failure mid-stream (connection reset, timeout) removes the temp file and
    leaves ``dest`` absent rather than truncated, so the next run downloads
    cleanly instead of tripping over a partial file.

    Args:
        response: Response from ``client.stream(...)``, used inside its context
            manager.
        dest: Destination file path.
        chunk_size: Read chunk size in bytes.
    """
    _atomic_write(dest, lambda tmp: _stream_chunks(response, tmp, chunk_size))

assert_zip_intact(path)

Verify that a ZIP archive is complete and all members pass CRC checks.

Parameters:

Name	Type	Description	Default
path	Path	Path to the ZIP file.	required

Raises:

Type	Description
BadZipFile	If the archive structure is corrupt or a member fails its CRC check.

Source code in pspcz_analyzer/fileio.py

def assert_zip_intact(path: Path) -> None:
    """Verify that a ZIP archive is complete and all members pass CRC checks.

    Args:
        path: Path to the ZIP file.

    Raises:
        zipfile.BadZipFile: If the archive structure is corrupt or a member
            fails its CRC check.
    """
    with zipfile.ZipFile(path) as zf:
        bad_member = zf.testzip()
    if bad_member is not None:
        raise zipfile.BadZipFile(f"Corrupt member in {path.name}: {bad_member}")