ifcquery/ifcedit: enable shell scripting by composing query and edit commands

Add --format ids to ifcquery to output step IDs suitable for piping into
ifcedit parameters. Add ifcedit foreach to apply an operation to every
element in a query result. Extend clash and relations output so --format ids
extracts all involved element IDs, enabling one-liners like clash detection
piped directly into render.

Generated with the assistance of an AI coding tool.

src/ifcedit/README.md

@@ -181,6 +181,50 @@ ifcedit run model.ifc pset.edit_pset --pset 15 \
     --properties '{"IsExternal": true, "FireRating": "2HR"}'
 ```
 
+### foreach
+
+Apply an API function to each element in a JSON array read from stdin.
+`{field}` placeholders in argument values are substituted with fields from
+each JSON object. The model is opened once and saved once regardless of how
+many elements are processed.
+
+```bash
+ifcquery model.ifc select 'IfcWindow' | ifcedit foreach model.ifc root.remove_product --product {id}
+```
+
+```json
+{"ok": true, "count": 36, "errors": []}
+```
+
+Placeholder tokens match the fields emitted by `ifcquery` — typically `{id}`,
+`{type}`, and `{name}`:
+
+```bash
+ifcquery model.ifc select 'IfcDoor' | ifcedit foreach model.ifc attribute.edit_attributes \
+    --product {id} --attributes '{"Name": "Door"}'
+```
+
+**Options:**
+
+- `-o, --output <path>` -- write to a different file instead of overwriting the input
+
+**Output:**
+
+- `count` -- number of elements successfully processed
+- `errors` -- list of per-element failures, each with `index`, `item`, and `error`; processing continues past errors
+
+```json
+{
+  "ok": false,
+  "count": 34,
+  "errors": [
+    {"index": 2, "item": {"id": 55, "type": "IfcWindow", "name": "W03"}, "error": "Entity #55 not found in model"}
+  ]
+}
+```
+
+Exit code is 1 if any element failed.
+
 ### quantify
 
 Run quantity take-off (QTO) on an IFC file, computing physical measurements
@@ -243,6 +287,19 @@ Exit code is 0 on success, 1 on error.
 A typical workflow: inspect with `ifcquery`, look up the right API function
 with `ifcedit docs`, then apply changes with `ifcedit run`.
 
+The two tools also compose directly in shell scripts. Use `ifcquery --format ids`
+to feed a list of IDs into a `run` parameter, or pipe `ifcquery select` JSON
+into `ifcedit foreach` to apply an operation to every matching element:
+
+```bash
+# Aggregate — pass all IDs as a list parameter
+ifcedit run model.ifc spatial.unassign_container \
+    --products "$(ifcquery model.ifc --format ids select 'IfcWall')"
+
+# Fan-out — one operation per element, model opened and saved once
+ifcquery model.ifc select 'IfcWindow' | ifcedit foreach model.ifc root.remove_product --product {id}
+```
+
 ## License
 
 LGPLv3+ -- see the IfcOpenShell project license.

src/ifcedit/ifcedit/__main__.py

@@ -28,6 +28,7 @@ import ifcopenshell
 from ifcedit.discover import function_docs, list_functions, list_modules
 from ifcedit.quantify import list_rules, run_quantify
 from ifcedit.run import run_api
+from ifcedit.foreach import run_foreach
 
 
 def format_output(data, fmt: str) -> str:
@@ -138,6 +139,42 @@ def _parse_extra_args(extra: list[str]) -> dict[str, str]:
     return kwargs
 
 
+def cmd_foreach(args, extra_args):
+    try:
+        model = ifcopenshell.open(args.ifc_file)
+    except Exception as e:
+        print(f"Error: Could not open IFC file: {e}", file=sys.stderr)
+        sys.exit(1)
+
+    parts = args.function_path.split(".")
+    if len(parts) != 2:
+        print("Error: function path must be 'module.function' (e.g. root.create_entity)", file=sys.stderr)
+        sys.exit(1)
+    module, function = parts
+
+    raw_kwargs_template = _parse_extra_args(extra_args)
+
+    try:
+        stdin_data = json.load(sys.stdin)
+    except json.JSONDecodeError as e:
+        print(f"Error: Could not parse JSON from stdin: {e}", file=sys.stderr)
+        sys.exit(1)
+
+    if not isinstance(stdin_data, list):
+        print("Error: stdin must be a JSON array", file=sys.stderr)
+        sys.exit(1)
+
+    result = run_foreach(model, module, function, raw_kwargs_template, stdin_data)
+
+    if result["ok"]:
+        output_path = args.output or args.ifc_file
+        model.write(output_path)
+
+    print(format_output(result, args.output_format))
+    if not result["ok"]:
+        sys.exit(1)
+
+
 def cmd_quantify(args, extra_args):
     if args.quantify_command == "list":
         result = list_rules()
@@ -191,6 +228,15 @@ def main():
     run_parser.add_argument("-o", "--output", help="Output file path (default: overwrite input)")
     run_parser.add_argument("--dry-run", action="store_true", help="Validate without executing or saving")
 
+    # foreach
+    foreach_parser = subparsers.add_parser(
+        "foreach",
+        help="Apply an API function to each element in a JSON array read from stdin",
+    )
+    foreach_parser.add_argument("ifc_file", help="Path to the IFC file")
+    foreach_parser.add_argument("function_path", help="module.function (e.g. attribute.edit_attributes)")
+    foreach_parser.add_argument("-o", "--output", help="Output file path (default: overwrite input)")
+
     # quantify
     quantify_parser = subparsers.add_parser("quantify", help="Quantity take-off (QTO) using ifc5d rules")
     quantify_sub = quantify_parser.add_subparsers(dest="quantify_command")
@@ -209,6 +255,8 @@ def main():
         cmd_docs(args)
     elif args.command == "run":
         cmd_run(args, extra)
+    elif args.command == "foreach":
+        cmd_foreach(args, extra)
     elif args.command == "quantify":
         cmd_quantify(args, extra)
 

src/ifcedit/ifcedit/foreach.py

@@ -0,0 +1,76 @@
+# IfcEdit - CLI wrapper for ifcopenshell.api mutation functions
+# Copyright (C) 2026 Bruno Postle <bruno@postle.net>
+#
+# This file is part of IfcEdit.
+#
+# IfcEdit is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# IfcEdit is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with IfcEdit.  If not, see <http://www.gnu.org/licenses/>.
+
+from __future__ import annotations
+
+import ifcopenshell
+
+from ifcedit.run import run_api
+
+
+def _substitute(template: str, item: dict) -> str:
+    """Replace {key} placeholders in template with values from item."""
+    for key, value in item.items():
+        template = template.replace(f"{{{key}}}", str(value))
+    return template
+
+
+def run_foreach(
+    model: ifcopenshell.file,
+    module: str,
+    function: str,
+    raw_kwargs_template: dict[str, str],
+    items: list[dict],
+) -> dict:
+    """Apply an API function to each item in a list, substituting {field} placeholders.
+
+    Opens the model once, applies the mutation for every item, and returns a summary.
+    The caller is responsible for saving the model.
+
+    Args:
+        model: The open IFC model (mutated in place).
+        module: API module name (e.g. "root").
+        function: Function name (e.g. "remove_product").
+        raw_kwargs_template: Arg templates with {field} placeholders, e.g. {"product": "{id}"}.
+        items: List of dicts (e.g. from ifcquery select output).
+
+    Returns:
+        {"ok": True, "count": N, "errors": []} on full success,
+        {"ok": False, "count": N, "errors": [{...}]} if any item failed.
+    """
+    errors = []
+    count = 0
+
+    for i, item in enumerate(items):
+        if not isinstance(item, dict):
+            errors.append({"index": i, "item": item, "error": "item is not a dict"})
+            continue
+
+        substituted = {k: _substitute(v, item) for k, v in raw_kwargs_template.items()}
+        result = run_api(model, module, function, substituted)
+
+        if result["ok"]:
+            count += 1
+        else:
+            errors.append({"index": i, "item": item, "error": result["error"]})
+
+    return {
+        "ok": len(errors) == 0,
+        "count": count,
+        "errors": errors,
+    }

src/ifcedit/tests/test_foreach.py

@@ -0,0 +1,79 @@
+# Tests for ifcedit.foreach
+import ifcopenshell
+import ifcopenshell.api.owner.settings
+import ifcopenshell.api.project
+import ifcopenshell.api.root
+import ifcopenshell.api.spatial
+import ifcopenshell.api.unit
+import pytest
+
+from ifcedit.foreach import _substitute, run_foreach
+
+
+@pytest.fixture
+def model(model):
+    return model
+
+
+class TestSubstitute:
+    def test_single_field(self):
+        assert _substitute("--product {id}", {"id": 42}) == "--product 42"
+
+    def test_multiple_fields(self):
+        result = _substitute("{type} #{id} ({name})", {"id": 5, "type": "IfcWall", "name": "W1"})
+        assert result == "IfcWall #5 (W1)"
+
+    def test_no_placeholder(self):
+        assert _substitute("hello", {"id": 1}) == "hello"
+
+    def test_unknown_placeholder_unchanged(self):
+        assert _substitute("{unknown}", {"id": 1}) == "{unknown}"
+
+
+class TestRunForeach:
+    def _items(self, model, ifc_class):
+        return [{"id": e.id(), "type": e.is_a(), "name": e.Name} for e in model.by_type(ifc_class)]
+
+    def test_rename_single(self, model):
+        items = self._items(model, "IfcWall")
+        result = run_foreach(model, "attribute", "edit_attributes", {"product": "{id}", "attributes": '{"Name": "R"}'}, items)
+        assert result["ok"] is True
+        assert result["count"] == 1
+        assert result["errors"] == []
+        assert model.by_type("IfcWall")[0].Name == "R"
+
+    def test_rename_multiple(self, model):
+        items = self._items(model, "IfcElement")
+        result = run_foreach(model, "attribute", "edit_attributes", {"product": "{id}", "attributes": '{"Name": "X"}'}, items)
+        assert result["ok"] is True
+        assert result["count"] == len(items)
+
+    def test_empty_list(self, model):
+        result = run_foreach(model, "root", "remove_product", {"product": "{id}"}, [])
+        assert result["ok"] is True
+        assert result["count"] == 0
+        assert result["errors"] == []
+
+    def test_bad_id_collects_error(self, model):
+        items = [{"id": 999999, "type": "IfcWall", "name": "X"}]
+        result = run_foreach(model, "root", "remove_product", {"product": "{id}"}, items)
+        assert result["ok"] is False
+        assert result["count"] == 0
+        assert len(result["errors"]) == 1
+        assert result["errors"][0]["index"] == 0
+
+    def test_non_dict_item_collects_error(self, model):
+        result = run_foreach(model, "root", "remove_product", {"product": "{id}"}, ["not_a_dict"])
+        assert result["ok"] is False
+        assert len(result["errors"]) == 1
+
+    def test_partial_failure_counts_successes(self, model):
+        wall_id = model.by_type("IfcWall")[0].id()
+        items = [
+            {"id": wall_id, "type": "IfcWall", "name": "W"},
+            {"id": 999999, "type": "IfcWall", "name": "Bad"},
+        ]
+        result = run_foreach(model, "attribute", "edit_attributes", {"product": "{id}", "attributes": '{"Name": "Ok"}'}, items)
+        assert result["ok"] is False
+        assert result["count"] == 1
+        assert len(result["errors"]) == 1

src/ifcedit/tests/test_main.py

@@ -3,15 +3,17 @@ import json
 import subprocess
 import sys
 
+import ifcopenshell
 import pytest
 
 
-def run_ifcedit(*args):
+def run_ifcedit(*args, stdin=None):
     """Run ifcedit as a subprocess and return (stdout, stderr, returncode)."""
     result = subprocess.run(
         [sys.executable, "-m", "ifcedit", *args],
         capture_output=True,
         text=True,
+        input=stdin,
     )
     return result.stdout, result.stderr, result.returncode
 
@@ -98,3 +100,82 @@ class TestRunCommand:
         stdout, stderr, rc = run_ifcedit("run", model_file, "invalid_path")
         assert rc != 0
         assert "module.function" in stderr
+
+
+class TestForeachCommand:
+    def _select_json(self, model, ifc_class):
+        """Build a JSON array like ifcquery select would produce."""
+        elements = model.by_type(ifc_class)
+        return json.dumps([{"id": e.id(), "type": e.is_a(), "name": getattr(e, "Name", None)} for e in elements])
+
+    def test_foreach_rename(self, model, model_file):
+        walls_json = self._select_json(model, "IfcWall")
+        stdout, stderr, rc = run_ifcedit(
+            "foreach", model_file, "attribute.edit_attributes",
+            "--product", "{id}", "--attributes", '{"Name": "Renamed"}',
+            stdin=walls_json,
+        )
+        assert rc == 0, f"stderr: {stderr}"
+        data = json.loads(stdout)
+        assert data["ok"] is True
+        assert data["count"] == 1
+        assert data["errors"] == []
+        updated = ifcopenshell.open(model_file)
+        assert updated.by_type("IfcWall")[0].Name == "Renamed"
+
+    def test_foreach_multiple_elements(self, model, model_file):
+        # Build a two-item list by selecting all IfcObject (includes spatial structure + elements)
+        elements_json = self._select_json(model, "IfcObject")
+        items = json.loads(elements_json)
+        assert len(items) >= 2
+        stdout, stderr, rc = run_ifcedit(
+            "foreach", model_file, "attribute.edit_attributes",
+            "--product", "{id}", "--attributes", '{"Name": "Bulk"}',
+            stdin=elements_json,
+        )
+        assert rc == 0, f"stderr: {stderr}"
+        data = json.loads(stdout)
+        assert data["ok"] is True
+        assert data["count"] == len(items)
+
+    def test_foreach_empty_list(self, model_file):
+        stdout, stderr, rc = run_ifcedit(
+            "foreach", model_file, "attribute.edit_attributes",
+            "--product", "{id}", "--attributes", '{"Name": "X"}',
+            stdin="[]",
+        )
+        assert rc == 0
+        data = json.loads(stdout)
+        assert data["ok"] is True
+        assert data["count"] == 0
+
+    def test_foreach_invalid_json_stdin(self, model_file):
+        stdout, stderr, rc = run_ifcedit(
+            "foreach", model_file, "root.remove_product", "--product", "{id}",
+            stdin="not json",
+        )
+        assert rc != 0
+        assert "Error" in stderr
+
+    def test_foreach_not_array_stdin(self, model_file):
+        stdout, stderr, rc = run_ifcedit(
+            "foreach", model_file, "root.remove_product", "--product", "{id}",
+            stdin='{"id": 1}',
+        )
+        assert rc != 0
+        assert "Error" in stderr
+
+    def test_foreach_output_to_different_file(self, model, model_file, tmp_path):
+        import os
+        output = str(tmp_path / "out.ifc")
+        walls_json = self._select_json(model, "IfcWall")
+        stdout, stderr, rc = run_ifcedit(
+            "foreach", model_file, "attribute.edit_attributes",
+            "-o", output,
+            "--product", "{id}", "--attributes", '{"Name": "OutFile"}',
+            stdin=walls_json,
+        )
+        assert rc == 0, f"stderr: {stderr}"
+        assert os.path.exists(output)
+        updated = ifcopenshell.open(output)
+        assert updated.by_type("IfcWall")[0].Name == "OutFile"