`elasticai.creator.ir`#

Package Contents#

Classes#

`AttributeMapping`
`DefaultDataGraphFactory`
`DefaultIrFactory`
`DefaultNodeEdgeFactory`
`DataGraphImpl`	Keep constructor signature when subclassing!
`Graph`
`GraphImpl`
`EdgeImpl`
`IrFactory`
`StdNodeEdgeFactory`
`NodeImpl`
`StdDataGraphFactory`
`StdIrFactory`
`DataGraph`	Note that the fact, that `add_node` and similar methods take an argument of type `Node` while the `nodes` and `edges` properties produce mappings over the generic type parameters `E` and `N`. This makes `DataGraph` covariant over `N` and `E`, meaning that `DataGraph[SpecialNode, SpecialEdge]` is a subtype of `DataGraph[Node, Edge]`. It also means that every implementation detail in a `DataGraph` implementation needs to assume that nodes and edges are of the most general `Node` or `Edge` type.
`Edge`
`Node`
`NodeEdgeFactory`
`ReadOnlyDataGraph`
`Registry`
`IrDeserializer`
`IrDeserializerLegacy`	Deserializes to the legacy format.
`IrSerializer`
`IrSerializerLegacy`	Serializer for the legacy format.

Functions#

attribute

Create AttributeMapping from other (native) data types recursively.

API#

class elasticai.creator.ir.AttributeMapping(**kwargs: elasticai.creator.ir._attribute.Attribute)[source]#

Bases: collections.abc.Mapping[str, elasticai.creator.ir._attribute.Attribute]

__getitem__(key: str) → Any[source]#

__iter__() → collections.abc.Iterator[str][source]#

__len__() → int[source]#

__eq__(other: object) → bool[source]#

__repr__() → str[source]#

drop(key: str) → Self[source]#

__or__(other: object) → Self[source]#

update_path(path: tuple[str, ...], value: elasticai.creator.ir._attribute.Attribute) → Self[source]#: update the entry found when following the path into nested Mappings

merge(other: collections.abc.Mapping) → Self[source]#

merge over nested mappings recursively

So instead of replacing a value found under a key, this checks wether that value is again an AttributeMapping and if so, updates it by the corresponding Mapping found in other. This happens recursively.

Opposed to that new_with and the | operator only allow to update values in the most outer AttributeMapping. Therefore using these to update a value in a nested structure, users would have to recreate the whole outer mapping structure until they reach the mapping they want to update.

Use Case:

If you would want to write

data = dict(a=dict(b=1, c=2))
data["a"]["b"] = 3

assert data["a"]["c"] == 2

You can instead

data = AttributeMapping(a=AttributeMapping(b=1, c=2))
data = data.merge(dict(a=dict(b=3)))
assert data["a"]["c"] == 2

If you want to update a single value, you can use the update_path() function instead to avoid having to build all the nested dictionaries.

new_with(**kwargs: elasticai.creator.ir._attribute.Attribute) → elasticai.creator.ir._attribute.AttributeMapping[source]#: replace key, value pairs in self by key, value pairs in kwargs

classmethod from_dict(data: dict) → elasticai.creator.ir._attribute.AttributeMapping[source]#

elasticai.creator.ir.attribute(arg: collections.abc.Mapping[str, elasticai.creator.ir._attribute.AttributeConvertable] | elasticai.creator.ir._attribute.AttributeMapping | elasticai.creator.ir._attribute.Attribute | None = None, /, **kwargs: elasticai.creator.ir._attribute.AttributeConvertable) → elasticai.creator.ir._attribute.AttributeMapping | tuple[elasticai.creator.ir._attribute.Attribute] | elasticai.creator.ir._attribute.Attribute[source]#

Create AttributeMapping from other (native) data types recursively.

The implementation assumes that any encountered AttributeMapping objects are correct, ie. they only contain Attributes themselves.

class elasticai.creator.ir.DefaultDataGraphFactory(node_edge_factory: elasticai.creator.ir.factories.NodeEdgeFactory[N, E])[source]#: Bases: elasticai.creator.ir.factories.StdDataGraphFactory[elasticai.creator.ir.datagraph_impl.DefaultDataGraphFactory.N, elasticai.creator.ir.datagraph_impl.DefaultDataGraphFactory.E, elasticai.creator.ir.datagraph.DataGraph]

class elasticai.creator.ir.DefaultIrFactory[source]#

Bases: elasticai.creator.ir.factories.StdIrFactory[elasticai.creator.ir.datagraph.Node, elasticai.creator.ir.datagraph.Edge, elasticai.creator.ir.datagraph.DataGraph[elasticai.creator.ir.datagraph.Node, elasticai.creator.ir.datagraph.Edge]]

Initialization

class elasticai.creator.ir.DefaultNodeEdgeFactory[source]#

Bases: elasticai.creator.ir.factories.NodeEdgeFactory[elasticai.creator.ir.datagraph.Node, elasticai.creator.ir.datagraph.Edge]

node(name: str, attributes: elasticai.creator.ir._attribute.AttributeMapping = AttributeMapping()) → elasticai.creator.ir.datagraph.Node[source]#

edge(src: str, dst: str, attributes: elasticai.creator.ir._attribute.AttributeMapping = AttributeMapping()) → elasticai.creator.ir.datagraph.Edge[source]#

class elasticai.creator.ir.DataGraphImpl(/, factory: elasticai.creator.ir._attribute.AttributeMapping, attributes: elasticai.creator.ir.graph.Graph[str, elasticai.creator.ir._attribute.AttributeMapping], graph: elasticai.creator.ir._attribute.AttributeMapping, node_attributes)[source]#

Bases: elasticai.creator.ir.datagraph.DataGraph[elasticai.creator.ir.datagraph_impl.DataGraphImpl.N, elasticai.creator.ir.datagraph_impl.DataGraphImpl.E]

Keep constructor signature when subclassing!

Initialization

__slots__#: (‘_factory’, ‘_graph’, ‘_attributes’, ‘_nodes’)

property node_attributes: elasticai.creator.ir._attribute.AttributeMapping#

property graph: elasticai.creator.ir.graph.Graph[str, elasticai.creator.ir._attribute.AttributeMapping]#

new_from_read_only_data_graph(g: elasticai.creator.ir.datagraph.ReadOnlyDataGraph) → Self[source]#

property attributes: elasticai.creator.ir._attribute.AttributeMapping#

property successors: collections.abc.Mapping[str, collections.abc.Mapping[str, elasticai.creator.ir._attribute.AttributeMapping]]#

property predecessors: collections.abc.Mapping[str, collections.abc.Mapping[str, elasticai.creator.ir._attribute.AttributeMapping]]#

property edges: collections.abc.Mapping[tuple[str, str], E]#

property nodes: collections.abc.Mapping[str, N]#

add_edge(src: str | elasticai.creator.ir.datagraph.Edge, dst: str | None = None, attributes: elasticai.creator.ir._attribute.AttributeMapping | None = None, /) → Self[source]#: Updates edge in case it exists already. Possibly already existing nodes remain unchanged.

add_edges(*edges: elasticai.creator.ir.datagraph.Edge | tuple[str, str, elasticai.creator.ir._attribute.AttributeMapping] | tuple[str, str]) → Self[source]#: Updates edges in case they exist already. Possibly already existing nodes remain unchanged.

add_node(name: str | elasticai.creator.ir.datagraph.Node, attributes: elasticai.creator.ir._attribute.AttributeMapping | None = None, /) → Self[source]#: Updates node in case it exists already.

add_nodes(*nodes: elasticai.creator.ir.datagraph.Node | tuple[str, elasticai.creator.ir._attribute.AttributeMapping] | str) → Self[source]#: Updates nodes in case they exist already.

remove_node(node: str, /) → Self[source]#: Will remove node and all connected edges.

remove_edge(src: str, dst: str) → Self[source]#: Will not remove nodes, even if they become isolated.

with_attributes(attributes: elasticai.creator.ir._attribute.AttributeMapping) → Self[source]#

property factory: elasticai.creator.ir.factories.NodeEdgeFactory[N, E]#

new(/, attributes: elasticai.creator.ir.graph.Graph[str, elasticai.creator.ir._attribute.AttributeMapping], graph: elasticai.creator.ir._attribute.AttributeMapping, node_attributes) → Self[source]#

__eq__(other: object) → bool[source]#

class elasticai.creator.ir.Graph[source]#

Bases: elasticai.creator.ir.graph.ReadOnlyGraph[elasticai.creator.ir.graph.Graph.N, elasticai.creator.ir.graph.Graph.E], typing.Protocol

abstractmethod add_node(node: N, /) → Self[source]#

abstractmethod add_edge(src: N, dst: N, attributes: E | None = None, /) → Self[source]#

abstractmethod remove_node(node: N, /) → Self[source]#

abstractmethod remove_edge(src: N, dst: N, /) → Self[source]#

abstractmethod add_nodes(*nodes: N) → Self[source]#

abstractmethod add_edges(*edges: tuple[N, N, E] | tuple[N, N]) → Self[source]#

class elasticai.creator.ir.GraphImpl(default_edge_attributes_factory: collections.abc.Callable[[], E], predecessors: elasticai.creator.ir.graph.AdjacencyMap[T, E] = AdjacencyMap(), successors: elasticai.creator.ir.graph.AdjacencyMap[T, E] = AdjacencyMap())[source]#

Bases: elasticai.creator.ir.graph.Graph[elasticai.creator.ir.graph.GraphImpl.T, elasticai.creator.ir.graph.GraphImpl.E]

add_node(node: T, /) → Self[source]#

add_nodes(*nodes: T) → Self[source]#

property successors: elasticai.creator.ir.graph.AdjacencyMap[T, E]#

property predecessors: elasticai.creator.ir.graph.AdjacencyMap[T, E]#

add_edge(src: T, dst: T, attributes: E | None = None, /) → Self[source]#

__eq__(other: object) → bool[source]#

add_edges(*edges: tuple[T, T, E | None] | tuple[T, T]) → Self[source]#

remove_edge(src: T, dst: T, /) → Self[source]#

remove_node(node: T, /) → Self[source]#

class elasticai.creator.ir.EdgeImpl(src: str, dst: str, attributes: elasticai.creator.ir._attribute.AttributeMapping)[source]#

Bases: elasticai.creator.ir.datagraph.Edge

__slots__#: (‘_src’, ‘_dst’, ‘_attributes’)

property attributes: elasticai.creator.ir._attribute.AttributeMapping#

property src: str#

property dst: str#

__hash__() → int[source]#

__eq__(other: object) → bool[source]#

__repr__() → str[source]#

class elasticai.creator.ir.IrFactory[source]#

Bases: elasticai.creator.ir.datagraph.NodeEdgeFactory[elasticai.creator.ir.factories.IrFactory.N, elasticai.creator.ir.factories.IrFactory.E], typing.Protocol

abstractmethod graph(attributes: elasticai.creator.ir._attribute.AttributeMapping = AttributeMapping(), /) → G[source]#

class elasticai.creator.ir.StdNodeEdgeFactory(node_fn: collections.abc.Callable[[str, elasticai.creator.ir._attribute.AttributeMapping], N], edge_fn: collections.abc.Callable[[str, str, elasticai.creator.ir._attribute.AttributeMapping], E])[source]#

Bases: elasticai.creator.ir.datagraph.NodeEdgeFactory

Initialization

node(name: str, attributes: elasticai.creator.ir._attribute.AttributeMapping = AttributeMapping()) → N[source]#

edge(src: str, dst: str, attributes: elasticai.creator.ir._attribute.AttributeMapping = AttributeMapping()) → E[source]#

class elasticai.creator.ir.NodeImpl(name: str, attributes: elasticai.creator.ir._attribute.AttributeMapping)[source]#

Bases: elasticai.creator.ir.datagraph.Node

__slots__#: (‘_name’, ‘_attributes’)

property name: str#

property type: str#

property attributes: elasticai.creator.ir._attribute.AttributeMapping#

__hash__() → int[source]#

__repr__() → str[source]#

__eq__(o: object) → bool[source]#

class elasticai.creator.ir.StdDataGraphFactory(node_edge: elasticai.creator.ir.datagraph.NodeEdgeFactory[N, E], graph_fn: collections.abc.Callable[[elasticai.creator.ir.datagraph.NodeEdgeFactory[N, E], elasticai.creator.ir._attribute.AttributeMapping], G])[source]#

Bases: elasticai.creator.ir.factories.DataGraphFactory[elasticai.creator.ir.factories.StdDataGraphFactory.G]

graph(attributes: elasticai.creator.ir._attribute.AttributeMapping = AttributeMapping()) → G[source]#

class elasticai.creator.ir.StdIrFactory(node_fn: collections.abc.Callable[[str, elasticai.creator.ir._attribute.AttributeMapping], N], edge_fn: collections.abc.Callable[[str, str, elasticai.creator.ir._attribute.AttributeMapping], E], graph_fn: collections.abc.Callable[[elasticai.creator.ir.datagraph.NodeEdgeFactory[N, E], elasticai.creator.ir._attribute.AttributeMapping], G])[source]#

Bases: elasticai.creator.ir.factories.StdNodeEdgeFactory[elasticai.creator.ir.factories.StdIrFactory.N, elasticai.creator.ir.factories.StdIrFactory.E]

Initialization

graph(attributes: elasticai.creator.ir._attribute.AttributeMapping = AttributeMapping()) → G[source]#

class elasticai.creator.ir.DataGraph[source]#

Bases: elasticai.creator.ir.datagraph.ReadOnlyDataGraph[elasticai.creator.ir.datagraph.N, elasticai.creator.ir.datagraph.E], elasticai.creator.ir.graph.Graph[str, elasticai.creator.ir._attribute.AttributeMapping], typing.Protocol[elasticai.creator.ir.datagraph.N, elasticai.creator.ir.datagraph.E]

Note that the fact, that add_node and similar methods take an argument of type Node while the nodes and edges properties produce mappings over the generic type parameters E and N. This makes DataGraph covariant over N and E, meaning that DataGraph[SpecialNode, SpecialEdge] is a subtype of DataGraph[Node, Edge]. It also means that every implementation detail in a DataGraph implementation needs to assume that nodes and edges are of the most general Node or Edge type.

abstractmethod add_nodes(*args: elasticai.creator.ir.datagraph.Node | tuple[str, elasticai.creator.ir._attribute.AttributeMapping] | str) → Self[source]#

abstractmethod add_edges(*args: elasticai.creator.ir.datagraph.Edge | tuple[str, str, elasticai.creator.ir._attribute.AttributeMapping] | tuple[str, str]) → Self[source]#

abstractmethod remove_node(node: str, /) → Self[source]#

abstractmethod remove_edge(src: str, dst: str) → Self[source]#

abstractmethod with_attributes(attributes: elasticai.creator.ir._attribute.AttributeMapping) → Self[source]#

class elasticai.creator.ir.Edge[source]#

Bases: typing.Protocol

abstract property attributes: elasticai.creator.ir._attribute.AttributeMapping#

abstract property src: str#

abstract property dst: str#

abstractmethod __eq__(o: object, /) → bool[source]#

class elasticai.creator.ir.Node[source]#

Bases: typing.Protocol

abstract property attributes: elasticai.creator.ir._attribute.AttributeMapping#

abstract property type: str#

abstract property name: str#

abstractmethod __eq__(o: object, /) → bool[source]#

class elasticai.creator.ir.NodeEdgeFactory[source]#

Bases: typing.Protocol[elasticai.creator.ir.datagraph.N, elasticai.creator.ir.datagraph.E]

abstractmethod node(name: str, attributes: elasticai.creator.ir._attribute.AttributeMapping = AttributeMapping(), /) → elasticai.creator.ir.datagraph.N[source]#

abstractmethod edge(src: str, dst: str, attributes: elasticai.creator.ir._attribute.AttributeMapping = AttributeMapping(), /) → elasticai.creator.ir.datagraph.E[source]#

class elasticai.creator.ir.ReadOnlyDataGraph[source]#

Bases: elasticai.creator.ir.graph.ReadOnlyGraph[str, elasticai.creator.ir._attribute.AttributeMapping], typing.Protocol[elasticai.creator.ir.datagraph.N, elasticai.creator.ir.datagraph.E]

abstract property attributes: elasticai.creator.ir._attribute.AttributeMapping#

abstract property nodes: collections.abc.Mapping[str, elasticai.creator.ir.datagraph.N]#

abstract property graph: elasticai.creator.ir.graph.Graph[str, elasticai.creator.ir._attribute.AttributeMapping]#

abstract property node_attributes: elasticai.creator.ir._attribute.AttributeMapping#

abstract property edges: collections.abc.Mapping[tuple[str, str], elasticai.creator.ir.datagraph.E]#

abstract property factory: elasticai.creator.ir.datagraph.NodeEdgeFactory[elasticai.creator.ir.datagraph.N, elasticai.creator.ir.datagraph.E]#

class elasticai.creator.ir.Registry(items: collections.abc.Iterable[tuple[str, G]] | collections.abc.Mapping[str, G] | None = None, /, **kwargs)[source]#

Bases: collections.abc.Mapping[str, elasticai.creator.ir.registry.Registry.G]

__getitem__(name: str) → G[source]#

__len__() → int[source]#

__contains__(key: object) → bool[source]#

__iter__() → collections.abc.Iterator[str][source]#

__or__(other: object) → Self[source]#

add(name: str, graph: G) → Self[source]#

class elasticai.creator.ir.IrDeserializer(factory: elasticai.creator.ir.factories.IrFactory[N, E, G])[source]#

Initialization

deserialize_graph(data: dict[str, Any]) → G[source]#

class elasticai.creator.ir.IrDeserializerLegacy(factory: elasticai.creator.ir.factories.IrFactory[N, E, G])[source]#

Deserializes to the legacy format.

The only difference is that attributes are not stored in a dedicated field but at the top-level dict. Use this if you need to load an IR from data that was serialized using the IrData data types.

Initialization

deserialize_graph(data: dict[str, Any]) → G[source]#

class elasticai.creator.ir.IrSerializer[source]#

serialize(item: elasticai.creator.ir._attribute.Attribute | elasticai.creator.ir.serializer._Node | elasticai.creator.ir.serializer._ReadOnlyDataGraph, /) → Any[source]#

class elasticai.creator.ir.IrSerializerLegacy[source]#

Serializer for the legacy format.

The only difference is that the attributes are not saved in a dedicated field, but instead directly in the top-level dict.

Use this if you need to store the data in a format that can be read by the old IrData based implementations.

Initialization

serialize(item: elasticai.creator.ir._attribute.Attribute | elasticai.creator.ir.serializer._Node | elasticai.creator.ir.serializer._ReadOnlyDataGraph, /) → Any[source]#

elasticai.creator.ir#

Package Contents#

Classes#

Functions#

API#

`elasticai.creator.ir`#