Skip to content

Commit

Permalink
started the refactor of MGraph
Browse files Browse the repository at this point in the history
  • Loading branch information
@DinisCruz
DinisCruz committed Jan 8, 2025
1 parent c48fd17 commit 07e2ab8
Show file tree
Hide file tree
Showing 7 changed files with 189 additions and 69 deletions.
53 changes: 46 additions & 7 deletions mgraph_ai/domain/MGraph.py
View file
Original file line number Diff line number Diff line change
@@ -1,14 +1,53 @@
from mgraph_ai.domain.MGraph__Graph import MGraph__Graph
from typing import Any, Dict, Type, List
from mgraph_ai.domain.MGraph__Edge import MGraph__Edge
from mgraph_ai.domain.MGraph__Node import MGraph__Node
from mgraph_ai.models.Model__MGraph__Edge import Model__MGraph__Edge
from mgraph_ai.models.Model__MGraph__Graph import Model__MGraph__Graph
from mgraph_ai.models.Model__MGraph__Node import Model__MGraph__Node
from mgraph_ai.schemas.Schema__MGraph__Attribute import Schema__MGraph__Attribute
from mgraph_ai.schemas.Schema__MGraph__Node import Schema__MGraph__Node
from osbot_utils.helpers.Random_Guid import Random_Guid
from osbot_utils.type_safe.Type_Safe import Type_Safe
from osbot_utils.type_safe.decorators.type_safe import type_safe

class MGraph(Type_Safe): # Main MGraph class that users will interact with
graph: Model__MGraph__Graph # Reference to the underlying graph model

class MGraph(Type_Safe):
graph: MGraph__Graph
@type_safe
def add_node(self, value : Any ,
node_type : Type[Schema__MGraph__Node ] = None,
attributes: Dict[Random_Guid, Schema__MGraph__Attribute] = None) -> MGraph__Node: # Add a new Node

def edges(self):
return self.graph.edges()
node = self.graph.new_node(value=value, node_type=node_type, attributes=attributes)
return self.wrap_node(node)

def nodes(self):
return self.graph.nodes()
@type_safe
def add_edge(self, from_node_id: Random_Guid,
to_node_id : Random_Guid) -> MGraph__Edge: # Add a new edge between nodes

edge = self.graph.new_edge(from_node_id=from_node_id, to_node_id=to_node_id)
return self.wrap_edge(edge)

def delete_node(self, node_id: Random_Guid) -> bool: # Remove a node and its connected edges
return self.graph.delete_node(node_id)

def delete_edge(self, edge_id: Random_Guid) -> bool: # Remove an edge
return self.graph.delete_edge(edge_id)

def node(self, node_id: Random_Guid) -> MGraph__Node: # Get a node by its ID
node = self.graph.get_node(node_id)
if node:
return MGraph__Node(node=Model__MGraph__Node(data=node), graph=self.graph)

def edge(self, edge_id: Random_Guid) -> MGraph__Edge: # Get an edge by its ID
edge = self.graph.get_edge(edge_id)
if edge:
return MGraph__Edge(edge=Model__MGraph__Edge(data=edge), graph=self.graph)

def edges(self) -> List[MGraph__Edge]: # Get all edges in the graph
return [self.wrap_edge(edge) for edge in self.graph.edges()]

def nodes(self) -> List[MGraph__Node]: # Get all nodes in the graph
return [self.wrap_node(node) for node in self.graph.nodes()]


27 changes: 21 additions & 6 deletions mgraph_ai/domain/MGraph__Attribute.py
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,22 @@
from mgraph_ai.models.Model__MGraph__Attribute import Model__MGraph__Attribute
from mgraph_ai.models.Model__MGraph__Graph import Model__MGraph__Graph
from osbot_utils.type_safe.Type_Safe import Type_Safe
from typing import Any
from osbot_utils.helpers.Random_Guid import Random_Guid
from mgraph_ai.models.Model__MGraph__Attribute import Model__MGraph__Attribute
from mgraph_ai.models.Model__MGraph__Graph import Model__MGraph__Graph
from osbot_utils.type_safe.Type_Safe import Type_Safe

class MGraph__Attribute(Type_Safe):
attribute: Model__MGraph__Attribute
graph : Model__MGraph__Graph
class MGraph__Attribute(Type_Safe): # Domain class for attributes
attribute: Model__MGraph__Attribute # Reference to attribute model
graph : Model__MGraph__Graph # Reference to graph model

def id(self) -> Random_Guid: # Get attribute ID
return self.attribute.data.attribute_id

def name(self) -> str: # Get attribute name
return str(self.attribute.data.attribute_name)

def value(self) -> Any: # Get attribute value
return self.attribute.value()

def set_value(self, value: Any) -> 'MGraph__Attribute': # Set attribute value with type checking
self.attribute.set_value(value)
return self
67 changes: 51 additions & 16 deletions mgraph_ai/domain/MGraph__Edge.py
View file
Original file line number Diff line number Diff line change
@@ -1,16 +1,51 @@
from mgraph_ai.domain.MGraph__Attribute import MGraph__Attribute
from mgraph_ai.models.Model__MGraph__Edge import Model__MGraph__Edge
from mgraph_ai.models.Model__MGraph__Graph import Model__MGraph__Graph
from osbot_utils.type_safe.Type_Safe import Type_Safe

class MGraph__Edge(Type_Safe):
edge : Model__MGraph__Edge
graph: Model__MGraph__Graph

def attribute(self, attribute_id):
attribute = self.node.attributes.get(attribute_id)
if attribute:
return MGraph__Attribute(attribute=attribute, graph=self.graph)

def attributes(self):
return [MGraph__Attribute(attribute=attribute, graph=self.graph) for attribute in self.edge.attributes.values()]
from typing import Any, List
from mgraph_ai.domain.MGraph__Node import MGraph__Node
from mgraph_ai.models.Model__MGraph__Attribute import Model__MGraph__Attribute
from mgraph_ai.models.Model__MGraph__Node import Model__MGraph__Node
from mgraph_ai.schemas.Schema__MGraph__Attribute import Schema__MGraph__Attribute
from osbot_utils.helpers.Random_Guid import Random_Guid
from mgraph_ai.domain.MGraph__Attribute import MGraph__Attribute
from mgraph_ai.models.Model__MGraph__Edge import Model__MGraph__Edge
from mgraph_ai.models.Model__MGraph__Graph import Model__MGraph__Graph
from osbot_utils.type_safe.Type_Safe import Type_Safe

class MGraph__Edge(Type_Safe): # Domain class for edges
edge : Model__MGraph__Edge # Reference to edge model
graph: Model__MGraph__Graph # Reference to graph model

def id(self) -> Random_Guid: # Get edge ID
return self.edge.data.edge_config.edge_id

def from_node(self) -> MGraph__Node: # Get source node
data = self.graph.get_node(self.edge.from_node_id())
if data:
return MGraph__Node(node = Model__MGraph__Node(data=data),
graph = self.graph )

def to_node(self) -> MGraph__Node: # Get target node
data = self.graph.get_node(self.edge.to_node_id())
if data:
return MGraph__Node(node = Model__MGraph__Node(data=data),
graph = self.graph )

def add_attribute(self, name : str ,
value : Any , # Add a new attribute to edge
attr_type: type = None) -> 'MGraph__Edge':

attribute = Schema__MGraph__Attribute(attribute_id = Random_Guid() ,
attribute_name = name ,
attribute_value = value ,
attribute_type = attr_type or type(value))
self.edge.add_attribute(attribute)
return self

def attribute(self, attribute_id: Random_Guid) -> MGraph__Attribute: # Get an attribute by ID
data = self.edge.get_attribute(attribute_id)
if data:
return MGraph__Attribute(attribute = Model__MGraph__Attribute(data=data),
graph = self.graph )

def attributes(self) -> List[MGraph__Attribute]: # Get all edge attributes
return [MGraph__Attribute(attribute = Model__MGraph__Attribute(data=attr),
graph = self.graph )
for attr in self.edge.data.attributes.values()]
10 changes: 5 additions & 5 deletions mgraph_ai/domain/MGraph__Graph.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,23 +8,23 @@
class MGraph__Graph(Type_Safe):
graph: Model__MGraph__Graph

def add_node(self, **kwargs):
def add_node(self, **kwargs)-> MGraph__Node:
node = self.model.add_node(**kwargs)
return MGraph__Node(node=node, graph=self)

def node(self, node_id):
node_model = self.nodes().get(node_id)
def node(self, node_id) -> MGraph__Node:
node_model = self.graph.data.nodes.get(node_id)
return MGraph__Node(node=node_model, graph=self)

def nodes(self):
return [self.node(node_id) for node_id in self.model.nodes.keys()]
return [MGraph__Node(node=node, graph=self) for node in self.graph.nodes()]

def edge(self, edge_id):
edge_model = self.graph.nodes().get(edge_id)
return MGraph__Edge(edge=edge_model, graph=self)

def edges(self):
return [self.edge(edge_id) for edge_id in self.model.edges.keys()]
return [MGraph__Edge(edge=edge, graph=self) for edge in self.graph.edges()]

def new_node(self):
node = Schema__MGraph__Node
Expand Down
54 changes: 41 additions & 13 deletions mgraph_ai/domain/MGraph__Node.py
View file
Original file line number Diff line number Diff line change
@@ -1,17 +1,45 @@
from mgraph_ai.domain.MGraph__Attribute import MGraph__Attribute
from mgraph_ai.models.Model__MGraph__Graph import Model__MGraph__Graph
from mgraph_ai.models.Model__MGraph__Node import Model__MGraph__Node
from osbot_utils.type_safe.Type_Safe import Type_Safe
from typing import Any, List
from mgraph_ai.models.Model__MGraph__Attribute import Model__MGraph__Attribute
from mgraph_ai.schemas.Schema__MGraph__Attribute import Schema__MGraph__Attribute
from osbot_utils.helpers.Random_Guid import Random_Guid
from mgraph_ai.domain.MGraph__Attribute import MGraph__Attribute
from mgraph_ai.models.Model__MGraph__Graph import Model__MGraph__Graph
from mgraph_ai.models.Model__MGraph__Node import Model__MGraph__Node
from osbot_utils.type_safe.Type_Safe import Type_Safe


class MGraph__Node(Type_Safe):
node : Model__MGraph__Node
graph: Model__MGraph__Graph
class MGraph__Node(Type_Safe): # Domain class for nodes
node : Model__MGraph__Node # Reference to node model
graph: Model__MGraph__Graph # Reference to graph model

def attribute(self, attribute_id):
attribute = self.node.attributes.get(attribute_id)
if attribute:
return MGraph__Attribute(attribute=attribute, graph=self.graph)
def value(self) -> Any: # Get node value
return self.node.value()

def attributes(self):
return [MGraph__Attribute(attribute=attribute, graph=self.graph) for attribute in self.node.attributes.values()]
def set_value(self, value: Any) -> 'MGraph__Node': # Set node value with type checking
self.node.set_value(value)
return self

def id(self) -> Random_Guid: # Get node ID
return self.node.data.node_config.node_id

def add_attribute(self, name : str ,
value : Any ,
attr_type: type = None) -> 'MGraph__Node': # Add a new attribute to node

attribute = Schema__MGraph__Attribute(attribute_id = Random_Guid() ,
attribute_name = name ,
attribute_value = value ,
attribute_type = attr_type or type(value))
self.node.add_attribute(attribute)
return self

def attribute(self, attribute_id: Random_Guid) -> MGraph__Attribute: # Get an attribute by ID
data = self.node.get_attribute(attribute_id)
if data:
return MGraph__Attribute(attribute = Model__MGraph__Attribute(data=data),
graph = self.graph )

def attributes(self) -> List[MGraph__Attribute]: # Get all node attributes
return [MGraph__Attribute(attribute = Model__MGraph__Attribute(data=attr),
graph = self.graph )
for attr in self.node.data.attributes.values()]
35 changes: 19 additions & 16 deletions mgraph_ai/models/Model__MGraph__Graph.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,21 @@
class Model__MGraph__Graph(Type_Safe):
data: Schema__MGraph__Graph

@type_safe
def add_node(self, node: Schema__MGraph__Node) -> Schema__MGraph__Node: # Add a node to the graph
self.data.nodes[node.node_config.node_id] = node
return node

@type_safe
def add_edge(self, edge: Schema__MGraph__Edge) -> Schema__MGraph__Edge: # Add an edge to the graph
if edge.from_node_id not in self.data.nodes:
raise ValueError(f"Source node {edge.from_node_id} not found")
if edge.to_node_id not in self.data.nodes:
raise ValueError(f"Target node {edge.to_node_id} not found")

self.data.edges[edge.edge_config.edge_id] = edge
return edge

@type_safe
def new_node(self, value: Any, node_type: type = None, attributes=None) -> Schema__MGraph__Node: # Create and add a new node to the graph
if node_type is None:
Expand Down Expand Up @@ -44,20 +59,8 @@ def new_edge(self, from_node_id: Random_Guid, to_node_id: Random_Guid) -> Schema

return self.add_edge(edge)

@type_safe
def add_node(self, node: Schema__MGraph__Node) -> Schema__MGraph__Node: # Add a node to the graph
self.data.nodes[node.node_config.node_id] = node
return node

@type_safe
def add_edge(self, edge: Schema__MGraph__Edge) -> Schema__MGraph__Edge: # Add an edge to the graph
if edge.from_node_id not in self.data.nodes:
raise ValueError(f"Source node {edge.from_node_id} not found")
if edge.to_node_id not in self.data.nodes:
raise ValueError(f"Target node {edge.to_node_id} not found")

self.data.edges[edge.edge_config.edge_id] = edge
return edge
def edges(self):
return self.data.edges.values()

def get_node(self, node_id: Random_Guid) -> Schema__MGraph__Node:
return self.data.nodes.get(node_id)
Expand All @@ -72,7 +75,7 @@ def nodes(self):
return self.data.nodes.values()

@type_safe
def remove_node(self, node_id: Random_Guid) -> 'Model__MGraph__Graph': # Remove a node and all its connected edges
def delete_node(self, node_id: Random_Guid) -> 'Model__MGraph__Graph': # Remove a node and all its connected edges
if node_id not in self.data.nodes:
return False

Expand All @@ -88,7 +91,7 @@ def remove_node(self, node_id: Random_Guid) -> 'Model__MGraph__Graph':
return True

@type_safe
def remove_edge(self, edge_id: Random_Guid) -> 'Model__MGraph__Graph': # Remove an edge from the graph
def delete_edge(self, edge_id: Random_Guid) -> 'Model__MGraph__Graph': # Remove an edge from the graph
if edge_id not in self.data.edges:
return False

Expand Down
Loading

0 comments on commit 07e2ab8

Please sign in to comment.