Skip to content

graflo.hq.bulk.encode

Rows on disk: typed identity keys, lossless values, compressed JSON Lines.

A staged or folded row is one JSON object per line, in zstd frames. JSON keeps an absent field apart from a null one. Values JSON cannot hold (dates, times, UUIDs, decimals) are written as {"$graflo": <type>, "v": <text>} and read back as the same Python value.

Attributes

TAG = '$graflo' module-attribute

__all__ = ['TAG', 'IdentityKeyError', 'decode', 'dumps', 'encode', 'identity_member', 'key_member', 'read_rows', 'typed_key', 'write_rows'] module-attribute

Classes

IdentityKeyError

Bases: ValueError

An identity value that cannot key a row: null, empty, NaN or not a scalar.

Source code in graflo/util/typed_values.py 
22
23
class IdentityKeyError(ValueError):
    """An identity value that cannot key a row: null, empty, NaN or not a scalar."""

Functions:

decode(value)

The inverse of :func:encode.

Source code in graflo/util/typed_values.py 
118
119
120
121
122
123
124
125
126
def decode(value: Any) -> Any:
    """The inverse of :func:`encode`."""
    if isinstance(value, dict):
        if len(value) == 2 and TAG in value and "v" in value:
            return _DECODERS[value[TAG]](value["v"])
        return {k: decode(v) for k, v in value.items()}
    if isinstance(value, list):
        return [decode(v) for v in value]
    return value

dumps(row)

One row as a JSON line.

Source code in graflo/hq/bulk/encode.py 
42
43
44
45
46
47
def dumps(row: dict[str, Any]) -> bytes:
    """One row as a JSON line."""
    return (
        json.dumps(row, separators=(",", ":"), ensure_ascii=False, allow_nan=False)
        + "\n"
    ).encode()

encode(value)

value as JSON-ready data, non-JSON scalars tagged.

Raises:

Type Description
TypeError

A value graflo does not write.
Source code in graflo/util/typed_values.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def encode(value: Any) -> Any:
    """*value* as JSON-ready data, non-JSON scalars tagged.

    Raises:
        TypeError: A value graflo does not write.
    """
    if value is None or isinstance(value, bool | int | str):
        return value
    if isinstance(value, float):
        return value if math.isfinite(value) else {TAG: "float", "v": repr(value)}
    if isinstance(value, dict):
        return {str(k): encode(v) for k, v in value.items()}
    if isinstance(value, list | tuple):
        return [encode(v) for v in value]
    if isinstance(value, dt.datetime):
        return {TAG: "datetime", "v": value.isoformat()}
    if isinstance(value, dt.date):
        return {TAG: "date", "v": value.isoformat()}
    if isinstance(value, dt.time):
        return {TAG: "time", "v": value.isoformat()}
    if isinstance(value, UUID):
        return {TAG: "uuid", "v": str(value)}
    if isinstance(value, Decimal):
        return {TAG: "decimal", "v": str(value)}
    raise TypeError(f"cannot stage a value of type {type(value).__name__}")

identity_member(value)

An edge identity field's value as part of the edge key; absent is null.

A value that cannot key a vertex keys an edge by its encoded form.

Source code in graflo/util/typed_values.py 
129
130
131
132
133
134
135
136
137
138
139
def identity_member(value: Any) -> Any:
    """An edge identity field's value as part of the edge key; absent is null.

    A value that cannot key a vertex keys an edge by its encoded form.
    """
    if value is None:
        return None
    try:
        return key_member(value)
    except IdentityKeyError:
        return encode(value)

key_member(value)

One identity value as [kind, value], compared as the targets compare it.

Numbers compare by value (1 and 1.0 are one key), booleans apart from numbers, strings apart from numbers (1 and "1" are two keys). A UUID keys as its text, the way every backend stores it.

Raises:

Type Description
IdentityKeyError

The value cannot key a row.
Source code in graflo/util/typed_values.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def key_member(value: Any) -> list[Any]:
    """One identity value as ``[kind, value]``, compared as the targets compare it.

    Numbers compare by value (``1`` and ``1.0`` are one key), booleans apart
    from numbers, strings apart from numbers (``1`` and ``"1"`` are two keys).
    A UUID keys as its text, the way every backend stores it.

    Raises:
        IdentityKeyError: The value cannot key a row.
    """
    if value is None or value == "":
        raise IdentityKeyError("empty identity value")
    if isinstance(value, bool):
        return ["b", value]
    if isinstance(value, int):
        return ["n", value]
    if isinstance(value, float | Decimal):
        if isinstance(value, float) and not math.isfinite(value):
            raise IdentityKeyError(f"identity value {value!r} is not finite")
        if value == int(value):
            return ["n", int(value)]
        return ["n", float(value)]
    if isinstance(value, str):
        return ["s", value]
    if isinstance(value, UUID):
        return ["s", str(value)]
    if isinstance(value, dt.datetime):
        return ["t", value.isoformat()]
    if isinstance(value, dt.date):
        return ["d", value.isoformat()]
    if isinstance(value, dt.time):
        return ["h", value.isoformat()]
    raise IdentityKeyError(
        f"identity value of type {type(value).__name__} is not a scalar"
    )

read_rows(path)

The rows of a zstd JSON Lines file, values decoded.

Source code in graflo/hq/bulk/encode.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def read_rows(path: Path) -> Iterator[dict[str, Any]]:
    """The rows of a zstd JSON Lines file, values decoded."""
    zstd = _deps.zstandard()
    with open(path, "rb") as raw:
        reader = zstd.ZstdDecompressor().stream_reader(raw, read_across_frames=True)
        buffer = b""
        while chunk := reader.read(1 << 20):
            buffer += chunk
            *lines, buffer = buffer.split(b"\n")
            for line in lines:
                if line:
                    yield decode(json.loads(line))
        if buffer.strip():
            yield decode(json.loads(buffer))

typed_key(values)

The canonical text of a tuple of identity values; equal keys, equal text.

A composite identity may lack members, as an upsert matches them: an absent one keys as null, an empty string as itself.

Raises:

Type Description
IdentityKeyError

No member has a value, or one cannot key a row.
Source code in graflo/util/typed_values.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def typed_key(values: Iterable[Any]) -> str:
    """The canonical text of a tuple of identity values; equal keys, equal text.

    A composite identity may lack members, as an upsert matches them: an
    absent one keys as null, an empty string as itself.

    Raises:
        IdentityKeyError: No member has a value, or one cannot key a row.
    """
    values = list(values)
    if all(v is None or v == "" for v in values):
        raise IdentityKeyError("no identity value")
    members = [
        None if v is None else ["s", ""] if v == "" else key_member(v) for v in values
    ]
    return json.dumps(members, separators=(",", ":"), ensure_ascii=False)

write_rows(path, rows)

Write JSON lines to path as one zstd frame, atomically; return the count.

The file appears complete or not at all, so a crash never leaves a part.

Source code in graflo/hq/bulk/encode.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def write_rows(path: Path, rows: Iterable[bytes]) -> int:
    """Write JSON lines to *path* as one zstd frame, atomically; return the count.

    The file appears complete or not at all, so a crash never leaves a part.
    """
    zstd = _deps.zstandard()
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + ".tmp")
    count = 0
    with open(tmp, "wb") as raw:
        with zstd.ZstdCompressor().stream_writer(raw, closefd=False) as out:
            for line in rows:
                out.write(line)
                count += 1
        raw.flush()
        os.fsync(raw.fileno())
    os.replace(tmp, path)
    return count