Skip to content

writer

Classes for building the Writer tree.

Constructing a writer tree from the schema makes it easy to decouple the writing implementation from the schema.

BinaryWriter dataclass

Bases: Writer

Variable byte length writer.

Source code in pyiceberg/avro/writer.py 
134
135
136
137
138
139
@dataclass(frozen=True)
class BinaryWriter(Writer):
    """Variable byte length writer."""

    def write(self, encoder: BinaryEncoder, val: Any) -> None:
        encoder.write_bytes(val)

DecimalWriter dataclass

Bases: Writer

Source code in pyiceberg/avro/writer.py 
142
143
144
145
146
147
148
149
150
151
152
@dataclass(frozen=True)
class DecimalWriter(Writer):
    precision: int = dataclassfield()
    scale: int = dataclassfield()

    def write(self, encoder: BinaryEncoder, val: Any) -> None:
        return encoder.write(decimal_to_bytes(val, byte_length=decimal_required_bytes(self.precision)))

    def __repr__(self) -> str:
        """Return string representation of this object."""
        return f"DecimalWriter({self.precision}, {self.scale})"

__repr__()

Return string representation of this object.

Source code in pyiceberg/avro/writer.py 
150
151
152
def __repr__(self) -> str:
    """Return string representation of this object."""
    return f"DecimalWriter({self.precision}, {self.scale})"

FixedWriter dataclass

Bases: Writer

Source code in pyiceberg/avro/writer.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@dataclass(frozen=True)
class FixedWriter(Writer):
    _len: int = dataclassfield()

    def write(self, encoder: BinaryEncoder, val: bytes) -> None:
        if len(val) != self._len:
            raise ValueError(f"Expected {self._len} bytes, got {len(val)}")
        encoder.write(val)

    def __len__(self) -> int:
        """Return the length of this object."""
        return self._len

    def __repr__(self) -> str:
        """Return string representation of this object."""
        return f"FixedWriter({self._len})"

__len__()

Return the length of this object.

Source code in pyiceberg/avro/writer.py 
125
126
127
def __len__(self) -> int:
    """Return the length of this object."""
    return self._len

__repr__()

Return string representation of this object.

Source code in pyiceberg/avro/writer.py 
129
130
131
def __repr__(self) -> str:
    """Return string representation of this object."""
    return f"FixedWriter({self._len})"

IntegerWriter dataclass

Bases: Writer

Longs and ints are encoded the same way, and there is no long in Python.

Source code in pyiceberg/avro/writer.py 
60
61
62
63
64
65
@dataclass(frozen=True)
class IntegerWriter(Writer):
    """Longs and ints are encoded the same way, and there is no long in Python."""

    def write(self, encoder: BinaryEncoder, val: int) -> None:
        encoder.write_int(val)

StructWriter dataclass

Bases: Writer

Source code in pyiceberg/avro/writer.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@dataclass(frozen=True)
class StructWriter(Writer):
    field_writers: Tuple[Tuple[Optional[int], Writer], ...] = dataclassfield()

    def write(self, encoder: BinaryEncoder, val: Record) -> None:
        for pos, writer in self.field_writers:
            # When pos is None, then it is a default value
            writer.write(encoder, val[pos] if pos is not None else None)

    def __eq__(self, other: Any) -> bool:
        """Implement the equality operator for this object."""
        return self.field_writers == other.field_writers if isinstance(other, StructWriter) else False

    def __repr__(self) -> str:
        """Return string representation of this object."""
        return f"StructWriter(tuple(({','.join(repr(field) for field in self.field_writers)})))"

    def __hash__(self) -> int:
        """Return the hash of the writer as hash of this object."""
        return hash(self.field_writers)

__eq__(other)

Implement the equality operator for this object.

Source code in pyiceberg/avro/writer.py 
176
177
178
def __eq__(self, other: Any) -> bool:
    """Implement the equality operator for this object."""
    return self.field_writers == other.field_writers if isinstance(other, StructWriter) else False

__hash__()

Return the hash of the writer as hash of this object.

Source code in pyiceberg/avro/writer.py 
184
185
186
def __hash__(self) -> int:
    """Return the hash of the writer as hash of this object."""
    return hash(self.field_writers)

__repr__()

Return string representation of this object.

Source code in pyiceberg/avro/writer.py 
180
181
182
def __repr__(self) -> str:
    """Return string representation of this object."""
    return f"StructWriter(tuple(({','.join(repr(field) for field in self.field_writers)})))"

Writer dataclass

Bases: Singleton

Source code in pyiceberg/avro/writer.py 
44
45
46
47
48
49
50
51
@dataclass(frozen=True)
class Writer(Singleton):
    @abstractmethod
    def write(self, encoder: BinaryEncoder, val: Any) -> Any: ...

    def __repr__(self) -> str:
        """Return string representation of this object."""
        return f"{self.__class__.__name__}()"

__repr__()

Return string representation of this object.

Source code in pyiceberg/avro/writer.py 
49
50
51
def __repr__(self) -> str:
    """Return string representation of this object."""
    return f"{self.__class__.__name__}()"