Base
Data models - Base Pydantic model with custom methods.
BaseCells (UserList, Generic)
Base abstract class for notebook cells.
Source code in databooks/data_models/base.py
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(self, **kwargs)
Return valid notebook cells from differences.
Source code in databooks/data_models/base.py
@abstractmethod
def resolve(self, **kwargs: Any) -> list:
"""Return valid notebook cells from differences."""
raise NotImplementedError
DatabooksBase (BaseModel)
Base Pydantic class with extras on managing fields.
Source code in databooks/data_models/base.py
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) != 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
fields_d[name] = (Any, self_val - other_val) # type: ignore
else:
fields_d[name] = (tuple, (self_val, other_val))
# Build Pydantic models dynamically
DiffInstance = create_model(
"Diff" + type(self).__name__,
__base__=type(self),
resolve=resolve,
is_diff=(bool, True),
**fields_d,
)
return cast(DiffModel, DiffInstance()) # it'll be filled in with the defaults
__str__(self)
special
Return outputs of repr.
Source code in databooks/data_models/base.py
def __str__(self) -> str:
"""Return outputs of __repr__."""
return repr(self)
__sub__(self, other)
special
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
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) != 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
fields_d[name] = (Any, self_val - other_val) # type: ignore
else:
fields_d[name] = (tuple, (self_val, other_val))
# Build Pydantic models dynamically
DiffInstance = create_model(
"Diff" + type(self).__name__,
__base__=type(self),
resolve=resolve,
is_diff=(bool, True),
**fields_d,
)
return cast(DiffModel, DiffInstance()) # it'll be filled in with the defaults
remove_fields(self, 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
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 (Protocol, Iterable, Generic)
Protocol for mypy static type checking.
Source code in databooks/data_models/base.py
class DiffModel(Protocol, Iterable):
"""Protocol for mypy static type checking."""
is_diff: bool
def resolve(self, *args: Any, **kwargs: Any) -> DatabooksBase:
"""Protocol method that returns a valid base object."""
resolve(self, *args, **kwargs)
Protocol method that returns a valid base object.
Source code in databooks/data_models/base.py
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 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 |
True |
Returns:
Type | Description |
---|---|
DatabooksBase | List[T] |
Model with selected fields from the differences |
Source code in databooks/data_models/base.py
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)