from __future__ import annotations
from typing import ClassVar
import pydantic
from flowrep import base_models
[docs]
class HandleModel(pydantic.BaseModel):
model_config = pydantic.ConfigDict(frozen=True)
node: base_models.Label | None
port: base_models.Label
delimiter: ClassVar[str] = "."
[docs]
@pydantic.model_serializer
def serialize(self) -> str:
if self.node is None:
return self.port
return self.delimiter.join([self.node, self.port])
[docs]
@pydantic.model_validator(mode="before")
@classmethod
def deserialize(cls, data):
if isinstance(data, str):
parts = data.split(".", 1)
if len(parts) == 1:
return {"node": None, "port": parts[0]}
return {"node": parts[0], "port": parts[1]}
return data
[docs]
class SourceHandle(HandleModel):
node: base_models.Label
port: base_models.Label
[docs]
class TargetHandle(HandleModel):
node: base_models.Label
port: base_models.Label
[docs]
class OutputTarget(HandleModel):
node: None = pydantic.Field(default=None, frozen=True)
port: base_models.Label
Edges = dict[TargetHandle, SourceHandle] # Communicate between siblings
InputEdges = dict[TargetHandle, InputSource] # Pass data into a subgraph
OutputEdges = dict[OutputTarget, SourceHandle | InputSource]