unpack_object
def unpack_object(obj: Union[Dict, BaseModel]) -> DictArguments:
obj- The object to unpack, generally a dict or BaseModel
Returns:
merge
@multimethod
def merge(a, b)The base merge function
Arguments:
a- The first object to mergeb- The second object to merge
Returns:
merge_atomic
@merge.register
def merge_atomic(a: Any, b: T) -> TMerge an atomic value with any other value
This function effectively handles cases where a value is being replaced.
For example, if the value integer 5 needed to be updated to 6, this function would handle that.
Arguments:
a- The first object to mergeb- The second object to merge
Returns:
The new value
merge_missing
@merge.register
def merge_missing(a: T, b: None) -> TMerge an object with a missing value
Arguments:
a- The first object to mergeb- The second object to merge
Returns:
The original object
merge_dict_item
@merge.register
def merge_dict_item(a: Dict, b: Dict) -> DictMerge two dictionaries
Arguments:
a- The first object to mergeb- The second object to merge
Returns:
The merged dictionary
merge_list
@merge.register
def merge_list(a: list, b: list) -> listMerge two lists
Arguments:
a- The first object to mergeb- The second object to merge
Returns:
A merged list
merge_tuple
@merge.register
def merge_tuple(a: tuple, b: tuple) -> tupleMerge two tuples
Arguments:
a- The first object to mergeb- The second object to merge
Returns:
A merged tuple
merge_set
@merge.register
def merge_set(a: set, b: set) -> setMerge two sets
Arguments:
a- The first object to mergeb- The second object to merge
Returns:
A merged set
merge_pydantic
@merge.register
def merge_pydantic(a: BaseModel, b: Any) -> BaseModelMerge a non-pydantic object into a pydantic model
Arguments:
a- The first object to mergeb- The second object to merge
Returns:
An updated pydantic model
merge_pydantic_right
@merge.register
def merge_pydantic_right(a: T, b: BaseModel) -> TMerge a pydantic model into a non-pydantic model
Arguments:
a- The first object to mergeb- The second object to merge
Returns:
An updated non-pydantic model
merge_models
def merge_models(a: T, b: T) -> TThis function is deprecated. Use merge instead
Arguments:
a- The first object to mergeb- The second object to merge
Returns:
An updated model
compute_graph_changes
def compute_graph_changes(
items: List[SpecProto], active_items: List[SpecProto]
) -> Tuple[List[SpecProto], List[SpecProto], List[SpecProto]]Computes a graph update for a list of items and the corresponding set of currently active items.
Arguments:
items- The new list of graph nodesactive_items- The current list of graph nodes. This does not have to be the full graph, only those relevant to the items set. Most commonly, these are drawn from the same data source.
Returns:
A three tuple of new items, updated items, and deleted items.