Create an edge from internal entry to `receiver`.
(self,
receiver: Node,
keys: dict[str,dict|str]|None=None)
| 201 | return self._exit.output.copy() |
| 202 | |
| 203 | def edge_from_entry(self, |
| 204 | receiver: Node, |
| 205 | keys: dict[str,dict|str]|None=None): |
| 206 | """Create an edge from internal entry to `receiver`.""" |
| 207 | if receiver is not self._exit: |
| 208 | receiver_node = self._nodes.get(receiver.name) |
| 209 | if receiver_node is None: |
| 210 | raise ValueError( |
| 211 | f"Receiver node '{receiver.name}' not found in graph '{self.name}'" |
| 212 | ) |
| 213 | if receiver_node is not receiver: |
| 214 | raise ValueError( |
| 215 | f"Receiver node instance mismatch for '{receiver.name}' in graph '{self.name}'" |
| 216 | ) |
| 217 | |
| 218 | for existing in self._entry.out_edges: |
| 219 | if existing._receiver is receiver: |
| 220 | raise ValueError( |
| 221 | f"duplicate edge: {self._entry.name} -> {receiver.name} already exists in graph '{self.name}'" |
| 222 | ) |
| 223 | |
| 224 | edge = Edge(self._entry, receiver, keys) |
| 225 | self._edges.append(edge) |
| 226 | self._entry.add_out_edge(edge) |
| 227 | receiver.add_in_edge(edge) |
| 228 | return edge |
| 229 | |
| 230 | def edge_to_exit(self, |
| 231 | sender: Node, |
no test coverage detected