some more for standard lib

src/wf_authoring/ops/__init__.py

@@ -15,14 +15,21 @@ from .sequences import (
 from .values import (
     CoalesceInput,
     ConstantInput,
+    MappingOutput,
+    PickPathInput,
     PickKeyInput,
+    ProjectFieldsInput,
+    RenameFieldsInput,
     RuntimeErrorInput,
     TruthyInput,
     ValueOutput,
     coalesce,
     constant,
     default_if_none,
+    pick_path,
     pick_key,
+    project_fields,
+    rename_fields,
     runtime_error,
     truthy,
 )
@@ -33,8 +40,12 @@ __all__ = [
     "ConstantInput",
     "CountOutput",
     "ItemOutput",
+    "MappingOutput",
     "MaybeItemOutput",
+    "PickPathInput",
     "PickKeyInput",
+    "ProjectFieldsInput",
+    "RenameFieldsInput",
     "RuntimeErrorInput",
     "SequenceInput",
     "TruthyInput",
@@ -49,7 +60,10 @@ __all__ = [
     "last_item",
     "last_item_or_none",
     "length",
+    "pick_path",
     "pick_key",
+    "project_fields",
+    "rename_fields",
     "runtime_error",
     "truthy",
 ]

src/wf_authoring/ops/values.py

@@ -33,6 +33,33 @@ class PickKeyInput(BaseModel):
     key: str
 
 
+class PickPathInput(BaseModel):
+    """Input model for selecting a nested mapping value by dotted path."""
+
+    mapping: dict[str, Any]
+    path: str
+
+
+class MappingOutput(BaseModel):
+    """Output model for ops that emit a mapping."""
+
+    mapping: dict[str, Any]
+
+
+class ProjectFieldsInput(BaseModel):
+    """Input model for selecting named fields from a mapping."""
+
+    mapping: dict[str, Any]
+    fields: list[str]
+
+
+class RenameFieldsInput(BaseModel):
+    """Input model for remapping existing mapping keys."""
+
+    mapping: dict[str, Any]
+    renames: dict[str, str]
+
+
 class TruthyInput(BaseModel):
     """Input model for routing by Python truthiness."""
 
@@ -87,6 +114,54 @@ def pick_key(input: PickKeyInput) -> ValueOutput:
     return ValueOutput(value=input.mapping.get(input.key))
 
 
+@node(
+    name="authoring.pick_path",
+    input_model=PickPathInput,
+    output_model=ValueOutput,
+    description="Select a nested mapping value by dotted path, returning None if missing.",
+)
+def pick_path(input: PickPathInput) -> ValueOutput:
+    """Select a nested mapping value by dotted path, returning None if missing."""
+    current: Any = input.mapping
+    for part in input.path.split("."):
+        if not isinstance(current, dict):
+            return ValueOutput(value=None)
+        current = current.get(part)
+        if current is None:
+            return ValueOutput(value=None)
+    return ValueOutput(value=current)
+
+
+@node(
+    name="authoring.project_fields",
+    input_model=ProjectFieldsInput,
+    output_model=MappingOutput,
+    description="Return only the requested existing fields from a mapping.",
+)
+def project_fields(input: ProjectFieldsInput) -> MappingOutput:
+    """Return only the requested existing fields from a mapping."""
+    return MappingOutput(
+        mapping={field: input.mapping[field] for field in input.fields if field in input.mapping}
+    )
+
+
+@node(
+    name="authoring.rename_fields",
+    input_model=RenameFieldsInput,
+    output_model=MappingOutput,
+    description="Rename existing mapping fields and omit missing source keys.",
+)
+def rename_fields(input: RenameFieldsInput) -> MappingOutput:
+    """Rename existing mapping fields and omit missing source keys."""
+    return MappingOutput(
+        mapping={
+            target: input.mapping[source]
+            for source, target in input.renames.items()
+            if source in input.mapping
+        }
+    )
+
+
 @node(
     name="authoring.truthy",
     input_model=TruthyInput,