jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

Update type hints in ContractEdge class

Browse source
This commit is contained in:
anovazzi1 2024-01-19 22:03:03 -03:00
commit 8ab47639bf
1 changed files with 3 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/backend/langflow/graph/edge/base.py
View file

@ -1,4 +1,4 @@
from typing import TYPE_CHECKING, List, Optional from typing import TYPE_CHECKING, Any, List, Optional
from loguru import logger from loguru import logger
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
@ -104,7 +104,7 @@ class ContractEdge(Edge):
self.is_fulfilled = False # Whether the contract has been fulfilled. self.is_fulfilled = False # Whether the contract has been fulfilled.
self.result: Any = None self.result: Any = None
def honor(self, source, target) -> None: def honor(self, source: "Vertex", target: "Vertex") -> None:
""" """
Fulfills the contract by setting the result of the source vertex to the target vertex's parameter. Fulfills the contract by setting the result of the source vertex to the target vertex's parameter.
If the edge is runnable, the source vertex is run with the message text and the target vertex's If the edge is runnable, the source vertex is run with the message text and the target vertex's
@ -142,7 +142,7 @@ class ContractEdge(Edge):
params[key] = [item for item in value if isinstance(item, (str, int, bool, float, list, dict))] params[key] = [item for item in value if isinstance(item, (str, int, bool, float, list, dict))]
return params return params
async def get_result(self, source, target): async def get_result(self, source: "Vertex", target: "Vertex"):
# Fulfill the contract if it has not been fulfilled. # Fulfill the contract if it has not been fulfilled.
if not self.is_fulfilled: if not self.is_fulfilled:
self.honor(source, target) self.honor(source, target)