youngkingdom/dify
1
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-04-25 21:26:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(api): support variable reference and substitution in Email delivery

Browse Source 
The EmailDeliveryConfig.body now support referencing variables generated
by precedent nodes.
This commit is contained in:
QuantumGhost
2026-01-21 15:19:17 +08:00
parent 32dab780ba
commit 20be1dd819
7 changed files with 189 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
api/core/workflow/nodes/human_input/entities.py
View File

@ -13,6 +13,7 @@ from pydantic import BaseModel, Field, field_validator, model_validator
from core.variables.consts import SELECTORS_LENGTH
from core.workflow.nodes.base import BaseNodeData
from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
from core.workflow.runtime import VariablePool
from .enums import ButtonStyle, DeliveryMethodType, EmailRecipientType, FormInputType, PlaceholderType, TimeoutUnit
@ -81,7 +82,21 @@ class EmailDeliveryConfig(BaseModel):
def body_with_url(self, url: str | None) -> str:
"""Return body content with url placeholder replaced."""
return self.replace_url_placeholder(self.body, url)
return self.render_body_template(body=self.body, url=url)
@classmethod
def render_body_template(
cls,
*,
body: str,
url: str | None,
variable_pool: VariablePool | None = None,
) -> str:
"""Render email body by replacing placeholders with runtime values."""
templated_body = cls.replace_url_placeholder(body, url)
if variable_pool is None:
return templated_body
return variable_pool.convert_template(templated_body).text
class _DeliveryMethodBase(BaseModel):
@ -90,6 +105,9 @@ class _DeliveryMethodBase(BaseModel):
enabled: bool = True
id: uuid.UUID = Field(default_factory=uuid.uuid4)
def extract_variable_selectors(self) -> Sequence[Sequence[str]]:
return ()
class WebAppDeliveryMethod(_DeliveryMethodBase):
"""Webapp delivery method configuration."""
@ -105,6 +123,16 @@ class EmailDeliveryMethod(_DeliveryMethodBase):
type: Literal[DeliveryMethodType.EMAIL] = DeliveryMethodType.EMAIL
config: EmailDeliveryConfig
def extract_variable_selectors(self) -> Sequence[Sequence[str]]:
variable_template_parser = VariableTemplateParser(template=self.config.body)
selectors: list[Sequence[str]] = []
for variable_selector in variable_template_parser.extract_variable_selectors():
value_selector = list(variable_selector.value_selector)
if len(value_selector) < SELECTORS_LENGTH:
continue
selectors.append(value_selector[:SELECTORS_LENGTH])
return selectors
DeliveryChannelConfig = Annotated[WebAppDeliveryMethod | EmailDeliveryMethod, Field(discriminator="type")]
@ -239,13 +267,23 @@ class HumanInputNodeData(BaseNodeData):
return field_names
def extract_variable_selector_to_variable_mapping(self, node_id: str) -> Mapping[str, Sequence[str]]:
variable_selectors = []
variable_template_parser = VariableTemplateParser(template=self.form_content)
variable_selectors.extend(variable_template_parser.extract_variable_selectors())
variable_mappings = {}
for variable_selector in variable_selectors:
qualified_variable_mapping_key = f"{node_id}.{variable_selector.variable}"
variable_mappings[qualified_variable_mapping_key] = variable_selector.value_selector
variable_mappings: dict[str, Sequence[str]] = {}
def _add_variable_selectors(selectors: Sequence[Sequence[str]]) -> None:
for selector in selectors:
if len(selector) < SELECTORS_LENGTH:
continue
qualified_variable_mapping_key = f"{node_id}.#{'.'.join(selector[:SELECTORS_LENGTH])}#"
variable_mappings[qualified_variable_mapping_key] = list(selector[:SELECTORS_LENGTH])
form_template_parser = VariableTemplateParser(template=self.form_content)
_add_variable_selectors(
[selector.value_selector for selector in form_template_parser.extract_variable_selectors()]
)
for delivery_method in self.delivery_methods:
if not delivery_method.enabled:
continue
_add_variable_selectors(delivery_method.extract_variable_selectors())
for input in self.inputs:
placeholder = input.placeholder

9
api/services/human_input_delivery_test_service.py
View File

@ -15,6 +15,7 @@ from core.workflow.nodes.human_input.entities import (
ExternalRecipient,
MemberRecipient,
)
from core.workflow.runtime import VariablePool
from extensions.ext_database import db
from extensions.ext_mail import mail
from libs.email_template_renderer import render_email_template
@ -42,6 +43,7 @@ class DeliveryTestContext:
rendered_content: str
template_vars: dict[str, str] = field(default_factory=dict)
recipients: list[DeliveryTestEmailRecipient] = field(default_factory=list)
variable_pool: VariablePool | None = None
@dataclass(frozen=True)
@ -154,9 +156,10 @@ class EmailDeliveryTestHandler:
recipient_email=recipient_email,
)
subject = render_email_template(method.config.subject, substitutions)
templated_body = EmailDeliveryConfig.replace_url_placeholder(
method.config.body,
substitutions.get("form_link"),
templated_body = EmailDeliveryConfig.render_body_template(
body=method.config.body,
url=substitutions.get("form_link"),
variable_pool=context.variable_pool,
)
body = render_email_template(templated_body, substitutions)

3
api/services/workflow_service.py
View File

@ -962,6 +962,7 @@ class WorkflowService:
rendered_content=rendered_content,
template_vars={"form_id": form_id},
recipients=recipients,
variable_pool=variable_pool,
)
try:
test_service.send_test(context=context, method=delivery_method)
@ -1087,8 +1088,6 @@ class WorkflowService:
config=node_config,
)
normalized_user_inputs: dict[str, Any] = dict(manual_inputs)
for raw_key, value in manual_inputs.items():
normalized_user_inputs[f"#{raw_key}#"] = value
load_into_variable_pool(
variable_loader=variable_loader,

41
api/tasks/mail_human_input_delivery_task.py
View File

@ -11,6 +11,8 @@ from sqlalchemy.orm import Session, sessionmaker
from configs import dify_config
from core.workflow.nodes.human_input.entities import EmailDeliveryConfig, EmailDeliveryMethod
from core.workflow.runtime import GraphRuntimeState, VariablePool
from core.app.layers.pause_state_persist_layer import WorkflowResumptionContext
from extensions.ext_database import db
from extensions.ext_mail import mail
from libs.email_template_renderer import render_email_template
@ -21,6 +23,7 @@ from models.human_input import (
HumanInputFormRecipient,
RecipientType,
)
from repositories.factory import DifyAPIRepositoryFactory
from services.feature_service import FeatureService
logger = logging.getLogger(__name__)
@ -108,11 +111,41 @@ def _render_subject(subject_template: str, substitutions: dict[str, str]) -> str
return render_email_template(subject_template, substitutions)
def _render_body(body_template: str, substitutions: dict[str, str]) -> str:
templated_body = EmailDeliveryConfig.replace_url_placeholder(body_template, substitutions.get("form_link"))
def _render_body(
body_template: str,
substitutions: dict[str, str],
*,
variable_pool: VariablePool | None,
) -> str:
templated_body = EmailDeliveryConfig.render_body_template(
body=body_template,
url=substitutions.get("form_link"),
variable_pool=variable_pool,
)
return render_email_template(templated_body, substitutions)
def _load_variable_pool(workflow_run_id: str | None) -> VariablePool | None:
if not workflow_run_id:
return None
session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_factory)
pause_entity = workflow_run_repo.get_workflow_pause(workflow_run_id)
if pause_entity is None:
logger.info("No pause state found for workflow run %s", workflow_run_id)
return None
try:
resumption_context = WorkflowResumptionContext.loads(pause_entity.get_state().decode())
except Exception:
logger.exception("Failed to load resumption context for workflow run %s", workflow_run_id)
return None
graph_runtime_state = GraphRuntimeState.from_snapshot(resumption_context.serialized_graph_runtime_state)
return graph_runtime_state.variable_pool
def _build_substitutions(
*,
job: _EmailDeliveryJob,
@ -163,11 +196,13 @@ def dispatch_human_input_email_task(form_id: str, node_title: str | None = None,
return
jobs = _load_email_jobs(session, form)
variable_pool = _load_variable_pool(form.workflow_run_id)
for job in jobs:
for recipient in job.recipients:
substitutions = _build_substitutions(job=job, recipient=recipient, node_title=node_title)
subject = _render_subject(job.subject, substitutions)
body = _render_body(job.body, substitutions)
body = _render_body(job.body, substitutions, variable_pool=variable_pool)
mail.send(
to=recipient.email,

15
api/tests/unit_tests/core/workflow/nodes/human_input/test_email_delivery_config.py
View File

@ -1,4 +1,5 @@
from core.workflow.nodes.human_input.entities import EmailDeliveryConfig, EmailRecipients
from core.workflow.runtime import VariablePool
def test_replace_url_placeholder_with_value():
@ -23,3 +24,17 @@ def test_replace_url_placeholder_missing_value():
result = config.body_with_url(None)
assert result == "No link available."
def test_render_body_template_replaces_variable_values():
config = EmailDeliveryConfig(
recipients=EmailRecipients(),
subject="Subject",
body="Hello {{#node1.value#}} {{#url#}}",
)
variable_pool = VariablePool()
variable_pool.add(["node1", "value"], "World")
result = config.render_body_template(body=config.body, url="https://example.com", variable_pool=variable_pool)
assert result == "Hello World https://example.com"

47
api/tests/unit_tests/services/test_human_input_delivery_test_service.py
View File

@ -14,6 +14,7 @@ from services.human_input_delivery_test_service import (
DeliveryTestError,
EmailDeliveryTestHandler,
)
from core.workflow.runtime import VariablePool
def _make_email_method() -> EmailDeliveryMethod:
@ -48,3 +49,49 @@ def test_email_delivery_test_handler_rejects_when_feature_disabled(monkeypatch:
with pytest.raises(DeliveryTestError, match="Email delivery is not available"):
handler.send_test(context=context, method=method)
def test_email_delivery_test_handler_replaces_body_variables(monkeypatch: pytest.MonkeyPatch):
class DummyMail:
def __init__(self):
self.sent: list[dict[str, str]] = []
def is_inited(self) -> bool:
return True
def send(self, *, to: str, subject: str, html: str):
self.sent.append({"to": to, "subject": subject, "html": html})
mail = DummyMail()
monkeypatch.setattr(service_module, "mail", mail)
monkeypatch.setattr(service_module, "render_email_template", lambda template, _substitutions: template)
monkeypatch.setattr(
service_module.FeatureService,
"get_features",
lambda _tenant_id: SimpleNamespace(human_input_email_delivery_enabled=True),
)
handler = EmailDeliveryTestHandler(session_factory=object())
handler._resolve_recipients = lambda **_kwargs: ["tester@example.com"] # type: ignore[assignment]
method = EmailDeliveryMethod(
config=EmailDeliveryConfig(
recipients=EmailRecipients(whole_workspace=False, items=[ExternalRecipient(email="tester@example.com")]),
subject="Subject",
body="Value {{#node1.value#}}",
)
)
variable_pool = VariablePool()
variable_pool.add(["node1", "value"], "OK")
context = DeliveryTestContext(
tenant_id="tenant-1",
app_id="app-1",
node_id="node-1",
node_title="Human Input",
rendered_content="content",
variable_pool=variable_pool,
)
handler.send_test(context=context, method=method)
assert mail.sent[0]["html"] == "Value OK"

38
api/tests/unit_tests/tasks/test_mail_human_input_delivery_task.py
View File

@ -49,7 +49,7 @@ def _build_job(recipient_count: int = 1) -> task_module._EmailDeliveryJob:
def test_dispatch_human_input_email_task_sends_to_each_recipient(monkeypatch: pytest.MonkeyPatch):
mail = _DummyMail()
form = SimpleNamespace(id="form-1", tenant_id="tenant-1")
form = SimpleNamespace(id="form-1", tenant_id="tenant-1", workflow_run_id=None)
def fake_render(template: str, substitutions: dict[str, str]) -> str:
return template.replace("{{ form_token }}", substitutions["form_token"]).replace(
@ -79,7 +79,7 @@ def test_dispatch_human_input_email_task_sends_to_each_recipient(monkeypatch: py
def test_dispatch_human_input_email_task_skips_when_feature_disabled(monkeypatch: pytest.MonkeyPatch):
mail = _DummyMail()
form = SimpleNamespace(id="form-1", tenant_id="tenant-1")
form = SimpleNamespace(id="form-1", tenant_id="tenant-1", workflow_run_id=None)
monkeypatch.setattr(task_module, "mail", mail)
monkeypatch.setattr(
@ -96,3 +96,37 @@ def test_dispatch_human_input_email_task_skips_when_feature_disabled(monkeypatch
)
assert mail.sent == []
def test_dispatch_human_input_email_task_replaces_body_variables(monkeypatch: pytest.MonkeyPatch):
mail = _DummyMail()
form = SimpleNamespace(id="form-1", tenant_id="tenant-1", workflow_run_id="run-1")
job = task_module._EmailDeliveryJob(
form_id="form-1",
workflow_run_id="run-1",
subject="Subject",
body="Body {{#node1.value#}}",
form_content="content",
recipients=[task_module._EmailRecipient(email="user@example.com", token="token-1")],
)
variable_pool = task_module.VariablePool()
variable_pool.add(["node1", "value"], "OK")
monkeypatch.setattr(task_module, "mail", mail)
monkeypatch.setattr(task_module, "render_email_template", lambda template, _substitutions: template)
monkeypatch.setattr(
task_module.FeatureService,
"get_features",
lambda _tenant_id: SimpleNamespace(human_input_email_delivery_enabled=True),
)
monkeypatch.setattr(task_module, "_load_email_jobs", lambda _session, _form: [job])
monkeypatch.setattr(task_module, "_load_variable_pool", lambda _workflow_run_id: variable_pool)
task_module.dispatch_human_input_email_task(
form_id="form-1",
node_title="Approve",
session_factory=lambda: _DummySession(form),
)
assert mail.sent[0]["html"] == "Body OK"