Skip to content

typedef

FrozenDict

Bases: Dict[Any, Any]

Source code in pyiceberg/typedef.py 
45
46
47
48
49
50
51
class FrozenDict(Dict[Any, Any]):
    def __setitem__(self, instance: Any, value: Any) -> None:
        """Assign a value to a FrozenDict."""
        raise AttributeError("FrozenDict does not support assignment")

    def update(self, *args: Any, **kwargs: Any) -> None:
        raise AttributeError("FrozenDict does not support .update()")

__setitem__(instance, value)

Assign a value to a FrozenDict.

Source code in pyiceberg/typedef.py 
46
47
48
def __setitem__(self, instance: Any, value: Any) -> None:
    """Assign a value to a FrozenDict."""
    raise AttributeError("FrozenDict does not support assignment")

IcebergBaseModel

Bases: BaseModel

This class extends the Pydantic BaseModel to set default values by overriding them.

This is because we always want to set by_alias to True. In Python, the dash can't be used in variable names, and this is used throughout the Iceberg spec.

The same goes for exclude_none, if a field is None we want to omit it from serialization, for example, the doc attribute on the NestedField object. Default non-null values will be serialized.

This is recommended by Pydantic: https://pydantic-docs.helpmanual.io/usage/model_config/#change-behaviour-globally

Source code in pyiceberg/typedef.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class IcebergBaseModel(BaseModel):
    """
    This class extends the Pydantic BaseModel to set default values by overriding them.

    This is because we always want to set by_alias to True. In Python, the dash can't
    be used in variable names, and this is used throughout the Iceberg spec.

    The same goes for exclude_none, if a field is None we want to omit it from
    serialization, for example, the doc attribute on the NestedField object.
    Default non-null values will be serialized.

    This is recommended by Pydantic:
    https://pydantic-docs.helpmanual.io/usage/model_config/#change-behaviour-globally
    """

    model_config = ConfigDict(populate_by_name=True, frozen=True)

    def _exclude_private_properties(self, exclude: Optional[Set[str]] = None) -> Set[str]:
        # A small trick to exclude private properties. Properties are serialized by pydantic,
        # regardless if they start with an underscore.
        # This will look at the dict, and find the fields and exclude them
        return set.union(
            {field for field in self.__dict__ if field.startswith("_") and not field == "__root__"}, exclude or set()
        )

    def model_dump(
        self, exclude_none: bool = True, exclude: Optional[Set[str]] = None, by_alias: bool = True, **kwargs: Any
    ) -> Dict[str, Any]:
        return super().model_dump(
            exclude_none=exclude_none, exclude=self._exclude_private_properties(exclude), by_alias=by_alias, **kwargs
        )

    def model_dump_json(
        self, exclude_none: bool = True, exclude: Optional[Set[str]] = None, by_alias: bool = True, **kwargs: Any
    ) -> str:
        return super().model_dump_json(
            exclude_none=exclude_none, exclude=self._exclude_private_properties(exclude), by_alias=by_alias, **kwargs
        )

IcebergRootModel

Bases: RootModel[T], Generic[T]

This class extends the Pydantic BaseModel to set default values by overriding them.

This is because we always want to set by_alias to True. In Python, the dash can't be used in variable names, and this is used throughout the Iceberg spec.

The same goes for exclude_none, if a field is None we want to omit it from serialization, for example, the doc attribute on the NestedField object. Default non-null values will be serialized.

This is recommended by Pydantic: https://pydantic-docs.helpmanual.io/usage/model_config/#change-behaviour-globally

Source code in pyiceberg/typedef.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class IcebergRootModel(RootModel[T], Generic[T]):
    """
    This class extends the Pydantic BaseModel to set default values by overriding them.

    This is because we always want to set by_alias to True. In Python, the dash can't
    be used in variable names, and this is used throughout the Iceberg spec.

    The same goes for exclude_none, if a field is None we want to omit it from
    serialization, for example, the doc attribute on the NestedField object.
    Default non-null values will be serialized.

    This is recommended by Pydantic:
    https://pydantic-docs.helpmanual.io/usage/model_config/#change-behaviour-globally
    """

    model_config = ConfigDict(frozen=True)

KeyDefaultDict

Bases: Dict[K, V]

Source code in pyiceberg/typedef.py 
63
64
65
66
67
68
69
70
71
72
class KeyDefaultDict(Dict[K, V]):
    def __init__(self, default_factory: Callable[[K], V]):
        super().__init__()
        self.default_factory = default_factory

    def __missing__(self, key: K) -> V:
        """Define behavior if you access a non-existent key in a KeyDefaultDict."""
        val = self.default_factory(key)
        self[key] = val
        return val

__missing__(key)

Define behavior if you access a non-existent key in a KeyDefaultDict.

Source code in pyiceberg/typedef.py 
68
69
70
71
72
def __missing__(self, key: K) -> V:
    """Define behavior if you access a non-existent key in a KeyDefaultDict."""
    val = self.default_factory(key)
    self[key] = val
    return val

Record

Bases: StructProtocol

Source code in pyiceberg/typedef.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
class Record(StructProtocol):
    __slots__ = ("_position_to_field_name",)
    _position_to_field_name: Tuple[str, ...]

    def __init__(self, *data: Any, struct: Optional[StructType] = None, **named_data: Any) -> None:
        if struct is not None:
            self._position_to_field_name = _get_struct_fields(struct)
        elif named_data:
            # Order of named_data is preserved (PEP 468) so this can be used to generate the position dict
            self._position_to_field_name = tuple(named_data.keys())
        else:
            self._position_to_field_name = tuple(f"field{idx + 1}" for idx in range(len(data)))

        for idx, d in enumerate(data):
            self[idx] = d

        for field_name, d in named_data.items():
            self.__setattr__(field_name, d)

    def __setitem__(self, pos: int, value: Any) -> None:
        """Assign a value to a Record."""
        self.__setattr__(self._position_to_field_name[pos], value)

    def __getitem__(self, pos: int) -> Any:
        """Fetch a value from a Record."""
        return self.__getattribute__(self._position_to_field_name[pos])

    def __eq__(self, other: Any) -> bool:
        """Return the equality of two instances of the Record class."""
        if not isinstance(other, Record):
            return False
        return self.__dict__ == other.__dict__

    def __repr__(self) -> str:
        """Return the string representation of the Record class."""
        return f"{self.__class__.__name__}[{', '.join(f'{key}={repr(value)}' for key, value in self.__dict__.items() if not key.startswith('_'))}]"

    def record_fields(self) -> List[str]:
        """Return values of all the fields of the Record class except those specified in skip_fields."""
        return [self.__getattribute__(v) if hasattr(self, v) else None for v in self._position_to_field_name]

__eq__(other)

Return the equality of two instances of the Record class.

Source code in pyiceberg/typedef.py 
189
190
191
192
193
def __eq__(self, other: Any) -> bool:
    """Return the equality of two instances of the Record class."""
    if not isinstance(other, Record):
        return False
    return self.__dict__ == other.__dict__

__getitem__(pos)

Fetch a value from a Record.

Source code in pyiceberg/typedef.py 
185
186
187
def __getitem__(self, pos: int) -> Any:
    """Fetch a value from a Record."""
    return self.__getattribute__(self._position_to_field_name[pos])

__repr__()

Return the string representation of the Record class.

Source code in pyiceberg/typedef.py 
195
196
197
def __repr__(self) -> str:
    """Return the string representation of the Record class."""
    return f"{self.__class__.__name__}[{', '.join(f'{key}={repr(value)}' for key, value in self.__dict__.items() if not key.startswith('_'))}]"

__setitem__(pos, value)

Assign a value to a Record.

Source code in pyiceberg/typedef.py 
181
182
183
def __setitem__(self, pos: int, value: Any) -> None:
    """Assign a value to a Record."""
    self.__setattr__(self._position_to_field_name[pos], value)

record_fields()

Return values of all the fields of the Record class except those specified in skip_fields.

Source code in pyiceberg/typedef.py 
199
200
201
def record_fields(self) -> List[str]:
    """Return values of all the fields of the Record class except those specified in skip_fields."""
    return [self.__getattribute__(v) if hasattr(self, v) else None for v in self._position_to_field_name]

StructProtocol

Bases: Protocol

A generic protocol used by accessors to get and set at positions of an object.

Source code in pyiceberg/typedef.py 
83
84
85
86
87
88
89
90
91
92
93
@runtime_checkable
class StructProtocol(Protocol):  # pragma: no cover
    """A generic protocol used by accessors to get and set at positions of an object."""

    @abstractmethod
    def __getitem__(self, pos: int) -> Any:
        """Fetch a value from a StructProtocol."""

    @abstractmethod
    def __setitem__(self, pos: int, value: Any) -> None:
        """Assign a value to a StructProtocol."""

__getitem__(pos) abstractmethod

Fetch a value from a StructProtocol.

Source code in pyiceberg/typedef.py 
87
88
89
@abstractmethod
def __getitem__(self, pos: int) -> Any:
    """Fetch a value from a StructProtocol."""

__setitem__(pos, value) abstractmethod

Assign a value to a StructProtocol.

Source code in pyiceberg/typedef.py 
91
92
93
@abstractmethod
def __setitem__(self, pos: int, value: Any) -> None:
    """Assign a value to a StructProtocol."""