mirror of
https://github.com/hoshikawa2/first_contas.git
synced 2026-07-09 18:24:20 +00:00
85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
from agente_contas_tim.workflows.exceptions import WorkflowConfigurationError
|
|
|
|
NormalizeMode = Literal[
|
|
"upper",
|
|
"lower",
|
|
"strip",
|
|
"upper_strip",
|
|
"lower_strip",
|
|
]
|
|
|
|
|
|
class PauseExpectedInputDef(BaseModel):
|
|
key: str = Field(..., min_length=1)
|
|
allowed_values: list[str] = Field(default_factory=list)
|
|
normalize: NormalizeMode | None = None
|
|
|
|
|
|
class PauseDef(BaseModel):
|
|
enabled: bool = True
|
|
when: dict[str, Any] | None = None
|
|
return_from: str = Field(..., min_length=1)
|
|
expected_input: PauseExpectedInputDef
|
|
resume_from: str = Field(..., min_length=1)
|
|
|
|
|
|
class NodeDef(BaseModel):
|
|
id: str = Field(..., min_length=1)
|
|
action: str = Field(..., min_length=1)
|
|
input: dict[str, Any] = Field(default_factory=dict)
|
|
pause: PauseDef | None = None
|
|
|
|
|
|
class EdgeDef(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
source: str = Field(..., alias="from", min_length=1)
|
|
target: str = Field(..., alias="to", min_length=1)
|
|
when: dict[str, Any] | None = None
|
|
priority: int = 100
|
|
|
|
|
|
class WorkflowDef(BaseModel):
|
|
name: str = Field(..., min_length=1)
|
|
version: int = Field(..., ge=1)
|
|
start: str = Field(..., min_length=1)
|
|
nodes: list[NodeDef]
|
|
edges: list[EdgeDef]
|
|
|
|
@model_validator(mode="after")
|
|
def validate_graph(self) -> WorkflowDef:
|
|
node_ids = {node.id for node in self.nodes}
|
|
if self.start not in node_ids:
|
|
raise WorkflowConfigurationError(
|
|
f"start={self.start!r} não existe em nodes"
|
|
)
|
|
|
|
for edge in self.edges:
|
|
if edge.source not in node_ids:
|
|
raise WorkflowConfigurationError(
|
|
f"edge.from={edge.source!r} não existe em nodes"
|
|
)
|
|
if edge.target != "END" and edge.target not in node_ids:
|
|
raise WorkflowConfigurationError(
|
|
f"edge.to={edge.target!r} não existe em nodes"
|
|
)
|
|
|
|
for node in self.nodes:
|
|
if node.pause is None:
|
|
continue
|
|
if (
|
|
node.pause.resume_from != "END"
|
|
and node.pause.resume_from not in node_ids
|
|
):
|
|
raise WorkflowConfigurationError(
|
|
f"pause.resume_from={node.pause.resume_from!r} não existe em nodes"
|
|
)
|
|
|
|
return self
|