feat(oa): persist relatedApproval links with field_key on start

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-07-22 17:01:53 +08:00
1 parent f2bf26cd8b
commit 14f00d5d3a
3 files changed
+44 -5

No files matched your search

+18 -5
View File
@@ -1449,16 +1449,29 @@ def start_instance(
)
db.add(inst)
for to_id in related_instance_ids or []:
to_id = (to_id or "").strip()
if not to_id:
continue
from app.services.oa_related_approval import (
assert_linkable_instance,
merge_relation_targets,
)
targets = merge_relation_targets(schema, data, related_instance_ids)
for field_key, to_id in targets:
target = (
db.query(OaInstance)
.filter(
OaInstance.id == to_id,
OaInstance.tenant_no == tenant_no,
OaInstance.del_flag == 0,
)
.first()
)
assert_linkable_instance(target, user_id=initiator_user_id, tenant_no=tenant_no)
db.add(
OaRelation(
id=generate_uuid(),
from_instance_id=inst.id,
to_instance_id=to_id,
field_key=None,
field_key=field_key,
tenant_no=tenant_no,
order_num=0,
remark=None,
@@ -117,6 +117,24 @@ def extract_related_approval_links(
return links
def merge_relation_targets(
schema: Any,
form_data: Optional[Dict[str, Any]],
legacy_ids: Optional[List[str]] = None,
) -> List[Tuple[Optional[str], str]]:
"""Returns (field_key|None, to_id) — form fields first, then legacy without key."""
links = extract_related_approval_links(schema, form_data)
result: List[Tuple[Optional[str], str]] = [(f, i) for f, i in links]
seen_ids = {i for _, i in links}
for x in legacy_ids or []:
tid = str(x or "").strip()
if not tid or tid in seen_ids:
continue
result.append((None, tid))
seen_ids.add(tid)
return result
def assert_linkable_instance(inst: OaInstance, *, user_id: str, tenant_no: str) -> None:
if not inst or inst.del_flag != 0 or inst.tenant_no != tenant_no:
raise OaEngineError("关联的审批单不存在")
@@ -3,6 +3,7 @@ import unittest
from app.services.oa_related_approval import (
extract_related_approval_links,
flatten_form_fields_for_summary,
merge_relation_targets,
)
@@ -57,5 +58,12 @@ class TestValidateLinkable(unittest.TestCase):
assert_linkable_instance(inst, user_id="u1", tenant_no="t")
class TestMergeRelationTargets(unittest.TestCase):
def test_merge_prefers_form_fields(self):
schema = {"rule": [{"type": "relatedApproval", "field": "r1"}]}
m = merge_relation_targets(schema, {"r1": ["a"]}, ["a", "b"])
self.assertEqual(m, [("r1", "a"), (None, "b")])
if __name__ == "__main__":
unittest.main()