How to Create a Mock Iceberg Snapshot Manifest in Python

import json
from datetime import datetime, timezone


def create_mock_manifest(snapshot_id: int, file_paths: list[str]) -> dict:
    """Create a mock Iceberg snapshot manifest structure."""
    manifest_file = {
        "manifest_path": f"/warehouse/table/metadata/snap-{snapshot_id}-m0.avro",
        "manifest_length": 8192,
        "partition_spec_id": 0,
        "content": 0,
        "sequence_number": snapshot_id,
        "min_sequence_number": snapshot_id,
        "added_snapshot_id": snapshot_id,
        "added_files_count": len(file_paths),
        "existing_files_count": 0,
        "deleted_files_count": 0,
        "added_rows_count": sum(1 for _ in file_paths),
        "existing_rows_count": 0,
        "deleted_rows_count": 0,
        "partitions": [],
        "key_metadata": None,
    }

    entries = []
    for idx, path in enumerate(file_paths):
        entry = {
            "status": 1,
            "snapshot_id": snapshot_id,
            "data_sequence_number": snapshot_id,
            "file_sequence_number": snapshot_id,
            "file_path": path,
            "file_format": "PARQUET",
            "partition": {},
            "record_count": 1000,
            "file_size_in_bytes": 65536,
            "column_sizes": [{"column_id": 1, "size": 4096}],
            "value_counts": [{"column_id": 1, "count": 1000}],
            "null_value_counts": [{"column_id": 1, "count": 0}],
            "nan_value_counts": [{"column_id": 1, "count": 0}],
            "lower_bounds": [{"column_id": 1, "value": f"value_{idx}"}],
            "upper_bounds": [{"column_id": 1, "value": f"value_{idx}"}],
            "key_metadata": None,
            "split_offsets": [4, 8192],
            "equality_ids": None,
            "sort_order_id": 0,
        }
        entries.append(entry)

    manifest = {
        "snapshot_id": snapshot_id,
        "timestamp_ms": int(datetime.now(timezone.utc).timestamp() * 1000),
        "summary": {
            "operation": "append",
            "added-data-files": str(len(file_paths)),
            "added-records": str(len(file_paths) * 1000),
            "added-files-size": str(len(file_paths) * 65536),
        },
        "manifest_list": f"/warehouse/table/metadata/snap-{snapshot_id}.avro",
        "manifests": [manifest_file],
        "data_entries": entries,
    }
    return manifest


if __name__ == "__main__":
    mock = create_mock_manifest(
        snapshot_id=42,
        file_paths=[
            "/warehouse/table/data/00000-0.parquet",
            "/warehouse/table/data/00001-0.parquet",
        ],
    )
    print(json.dumps(mock, indent=2))