Skip to content

Base

Data models - Base Pydantic model with custom methods.

BaseCells

Bases: UserList, Generic[T]

Base abstract class for notebook cells.

Source code in databooks/data_models/base.py 
25
26
27
28
29
30
31
class BaseCells(UserList, Generic[T]):
    """Base abstract class for notebook cells."""

    @abstractmethod
    def resolve(self, **kwargs: Any) -> list:
        """Return valid notebook cells from differences."""
        raise NotImplementedError

resolve(**kwargs) abstractmethod

Return valid notebook cells from differences.

Source code in databooks/data_models/base.py 
28
29
30
31
@abstractmethod
def resolve(self, **kwargs: Any) -> list:
    """Return valid notebook cells from differences."""
    raise NotImplementedError

DatabooksBase

Bases: BaseModel

Base Pydantic class with extras on managing fields.

Source code in databooks/data_models/base.py 
 88
 89
 90
 91
 92
 93
 94
 95
 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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class DatabooksBase(BaseModel):
    """Base Pydantic class with extras on managing fields."""

    model_config = ConfigDict(extra="allow")

    def remove_fields(
        self,
        fields: Iterable[str],
        *,
        recursive: bool = False,
        missing_ok: bool = False,
    ) -> None:
        """
        Remove selected fields.

        :param fields: Fields to remove
        :param recursive: Whether to remove the fields recursively in case of nested
         models
        :param missing_ok: Whether to raise errors in case field is missing
        :return:
        """
        d_model = dict(self)
        for field in fields:
            field_val = d_model.get(field) if missing_ok else d_model[field]
            if recursive and isinstance(field_val, DatabooksBase):
                field_val.remove_fields(fields)
            elif field in d_model:
                delattr(self, field)

    def __str__(self) -> str:
        """Return outputs of __repr__."""
        return repr(self)

    def __sub__(self, other: DatabooksBase) -> DiffModel:
        """
        Subtraction between `databooks.data_models.base.DatabooksBase` objects.

        The difference basically return models that replace each fields by a tuple,
         where for each field we have `field = (self_value, other_value)`
        """
        if type(self) is not type(other):
            raise TypeError(
                f"Unsupported operand types for `-`: `{type(self).__name__}` and"
                f" `{type(other).__name__}`"
            )

        # Get field and values for each instance
        self_d = dict(self)
        other_d = dict(other)

        # Build dict with {field: (type, value)} for each field
        fields_d: Dict[str, Any] = {}
        for name in self_d.keys() | other_d.keys():
            self_val = self_d.get(name)
            other_val = other_d.get(name)
            if type(self_val) is type(other_val) and all(
                isinstance(val, (DatabooksBase, BaseCells))
                for val in (self_val, other_val)
            ):
                # Recursively get the diffs for nested models; the `all(...)` check
                # above does not narrow the values for the type checker
                diff = self_val - other_val  # ty: ignore[unsupported-operator]
                fields_d[name] = (Any, diff)
            else:
                fields_d[name] = (tuple, (self_val, other_val))

        # Build Pydantic models dynamically
        DiffInstance = create_model(
            "Diff" + type(self).__name__,
            __base__=type(self),
            __namespace__={"resolve": resolve},
            is_diff=(bool, True),
            **fields_d,
        )

        return cast(DiffModel, DiffInstance())  # it'll be filled in with the defaults

__str__()

Return outputs of repr.

Source code in databooks/data_models/base.py 
117
118
119
def __str__(self) -> str:
    """Return outputs of __repr__."""
    return repr(self)

__sub__(other)

Subtraction between databooks.data_models.base.DatabooksBase objects.

The difference basically return models that replace each fields by a tuple, where for each field we have field = (self_value, other_value)

Source code in databooks/data_models/base.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def __sub__(self, other: DatabooksBase) -> DiffModel:
    """
    Subtraction between `databooks.data_models.base.DatabooksBase` objects.

    The difference basically return models that replace each fields by a tuple,
     where for each field we have `field = (self_value, other_value)`
    """
    if type(self) is not type(other):
        raise TypeError(
            f"Unsupported operand types for `-`: `{type(self).__name__}` and"
            f" `{type(other).__name__}`"
        )

    # Get field and values for each instance
    self_d = dict(self)
    other_d = dict(other)

    # Build dict with {field: (type, value)} for each field
    fields_d: Dict[str, Any] = {}
    for name in self_d.keys() | other_d.keys():
        self_val = self_d.get(name)
        other_val = other_d.get(name)
        if type(self_val) is type(other_val) and all(
            isinstance(val, (DatabooksBase, BaseCells))
            for val in (self_val, other_val)
        ):
            # Recursively get the diffs for nested models; the `all(...)` check
            # above does not narrow the values for the type checker
            diff = self_val - other_val  # ty: ignore[unsupported-operator]
            fields_d[name] = (Any, diff)
        else:
            fields_d[name] = (tuple, (self_val, other_val))

    # Build Pydantic models dynamically
    DiffInstance = create_model(
        "Diff" + type(self).__name__,
        __base__=type(self),
        __namespace__={"resolve": resolve},
        is_diff=(bool, True),
        **fields_d,
    )

    return cast(DiffModel, DiffInstance())  # it'll be filled in with the defaults

remove_fields(fields, *, recursive=False, missing_ok=False)

Remove selected fields.

Parameters:

Name Type Description Default
fields Iterable[str]

Fields to remove

 required
recursive bool

Whether to remove the fields recursively in case of nested models

 False
missing_ok bool

Whether to raise errors in case field is missing

 False

Returns:

Type Description
None
Source code in databooks/data_models/base.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def remove_fields(
    self,
    fields: Iterable[str],
    *,
    recursive: bool = False,
    missing_ok: bool = False,
) -> None:
    """
    Remove selected fields.

    :param fields: Fields to remove
    :param recursive: Whether to remove the fields recursively in case of nested
     models
    :param missing_ok: Whether to raise errors in case field is missing
    :return:
    """
    d_model = dict(self)
    for field in fields:
        field_val = d_model.get(field) if missing_ok else d_model[field]
        if recursive and isinstance(field_val, DatabooksBase):
            field_val.remove_fields(fields)
        elif field in d_model:
            delattr(self, field)

DiffModel

Bases: Iterable, Protocol

Protocol for static type checking.

Source code in databooks/data_models/base.py 
15
16
17
18
19
20
21
22
@runtime_checkable
class DiffModel(Iterable, Protocol):
    """Protocol for static type checking."""

    is_diff: bool

    def resolve(self, *args: Any, **kwargs: Any) -> DatabooksBase:
        """Protocol method that returns a valid base object."""

resolve(*args, **kwargs)

Protocol method that returns a valid base object.

Source code in databooks/data_models/base.py 
21
22
def resolve(self, *args: Any, **kwargs: Any) -> DatabooksBase:
    """Protocol method that returns a valid base object."""

resolve(model, *, keep_first=True, ignore_none=True, **kwargs) 

resolve(model: DiffModel, **kwargs: Any) -> DatabooksBase

resolve(model: BaseCells, **kwargs: Any) -> List[T]

Resolve differences for 'diff models'.

Return instance alike the parent class databooks.data_models.base.DatabooksBase.

Parameters:

Name Type Description Default
model DiffModel | BaseCells

DiffModel that is to be resolved (self when added as a method to a class

 required
keep_first bool

Whether to keep the information from the prior in the 'diff model' or the latter

 True
ignore_none bool

Whether to ignore None values if encountered, and use the other field value

 True

Returns:

Type Description
DatabooksBase | List[T]

Model with selected fields from the differences
Source code in databooks/data_models/base.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def resolve(
    model: DiffModel | BaseCells,
    *,
    keep_first: bool = True,
    ignore_none: bool = True,
    **kwargs: Any,
) -> DatabooksBase | List[T]:
    """
    Resolve differences for 'diff models'.

    Return instance alike the parent class `databooks.data_models.base.DatabooksBase`.
    :param model: DiffModel that is to be resolved (self when added as a method to a
     class
    :param keep_first: Whether to keep the information from the prior in the
     'diff model' or the latter
    :param ignore_none: Whether to ignore `None` values if encountered, and use the
     other field value
    :return: Model with selected fields from the differences
    """
    field_d = dict(model)
    is_diff = field_d.pop("is_diff")
    if not is_diff:
        raise TypeError("Can only resolve dynamic 'diff models' (when `is_diff=True`).")

    res_vals: Dict[str, Any] = {}
    for name, value in field_d.items():
        if isinstance(value, (DiffModel, BaseCells)):
            res_vals[name] = value.resolve(
                keep_first=keep_first, ignore_none=ignore_none, **kwargs
            )
        else:
            res_vals[name] = (
                value[keep_first]
                if value[not keep_first] is None and ignore_none
                else value[not keep_first]
            )

    return type(model).mro()[1](**res_vals)