feat: Add Input Aggregator component template

components/dataset_manipulation/Aggregate_inputs/README.md

@@ -0,0 +1,65 @@
+# Input Aggregator
+
+Author: Morgan Wowk <morgan.wowk@gmail.com>
+
+Location: [GitHub](https://github.com/Ark-kun/pipeline_components/blob/master/components/dataset_manipulation/Aggregate_inputs/component.yaml), [Raw](https://raw.githubusercontent.com/Ark-kun/pipeline_components/master/components/dataset_manipulation/Aggregate_inputs/component.yaml)
+
+Aggregates multiple pipeline inputs into a single output. Supports JsonArray, JsonObject, and CSV aggregation modes. Pass values as `agg_N` inputs and set `output_type` to control the aggregation format.
+
+## Inputs
+
+|Name|Type|Default|Description|
+|-|-|-|-|
+|output_type|[String]|JsonArray|Output format: `JsonArray` (array of values), `JsonObject` (keyed by input name), or `CSV` (union of CSV inputs with matching columns).|
+|agg_1, agg_2, ...|[String]||Values to aggregate. Inputs are added dynamically at runtime — one per value to aggregate.|
+
+## Outputs
+
+|Name|Type|Description|
+|-|-|-|
+|Output|[String]|The aggregated result in the format specified by `output_type`.|
+
+## Implementation
+
+#### Container
+
+Container image: [python:3.12-slim](https://hub.docker.com/_/python)
+
+## Usage
+
+```python
+input_aggregator_op = components.load_component_from_url(
+    "https://raw.githubusercontent.com/Ark-kun/pipeline_components/master/components/dataset_manipulation/Aggregate_inputs/component.yaml"
+)
+...
+input_aggregator_task = input_aggregator_op(
+    output_type="JsonArray",
+    agg_1=...,
+    agg_2=...,
+)
+```
+
+### Output type examples
+
+**`JsonArray`** — produces an ordered array of input values:
+```json
+["value1", "value2", "value3"]
+```
+
+**`JsonObject`** — produces an object keyed by input name (`agg_1`, `agg_2`, ...):
+```json
+{"agg_1": "value1", "agg_2": "value2", "agg_3": "value3"}
+```
+
+**`CSV`** — unions multiple CSV inputs with matching columns (header row from first input, subsequent headers are discarded):
+```
+col_a,col_b
+row1a,row1b
+row2a,row2b
+```
+
+## Other information
+
+###### Annotations
+
+* `is_input_aggregator: "true"`

components/dataset_manipulation/Aggregate_inputs/component.py

@@ -0,0 +1,98 @@
+from cloud_pipelines.components import OutputPath, create_component_from_func
+
+
+def aggregate(
+    output_path: OutputPath("String"),
+    output_type: str = "JsonArray",
+    **input_values: str,
+) -> None:
+    """Aggregates multiple pipeline inputs into a single output.
+
+    Supports JsonArray, JsonObject, and CSV aggregation modes.
+    Pass values as agg_N inputs and set output_type to control the aggregation format.
+
+    Args:
+        output_path: Path to write the aggregated output.
+        output_type: Output format — JsonArray (array of values), JsonObject (keyed by
+            input name), or CSV (union of CSV inputs with matching columns).
+        **input_values: Dynamic agg_N inputs to aggregate.
+
+    Annotations:
+        author: Tangle Team
+    """
+    import csv
+    import io
+    import json
+    import os
+
+    inputs = {}
+    for key, value in sorted(input_values.items()):
+        if value is not None:
+            inputs[key] = value
+
+    if not inputs:
+        raise ValueError("No inputs provided. At least one input must be connected.")
+
+    if output_type == "JsonArray":
+        result = json.dumps(list(inputs.values()), indent=2)
+
+    elif output_type == "JsonObject":
+        result = json.dumps(inputs, indent=2)
+
+    elif output_type == "CSV":
+        all_rows = []
+        header = None
+
+        def _normalize_csv_rows(rows: list[list[str]]) -> list[list[str]]:
+            """Strip leading empty column (pandas index) from all rows if present."""
+            if rows and rows[0] and rows[0][0] == "":
+                return [row[1:] if row else row for row in rows]
+            return rows
+
+        for _name, csv_text in inputs.items():
+            text = csv_text.strip()
+            if not text:
+                continue
+            reader = csv.reader(io.StringIO(text))
+            rows = _normalize_csv_rows(list(reader))
+            if not rows or rows == [[""]]:
+                continue
+            if header is None:
+                header = rows[0]
+                all_rows.append(header)
+                all_rows.extend(rows[1:])
+            else:
+                # Always treat the first row of subsequent inputs as a header and skip it.
+                # Raise only if column count differs, which would produce malformed CSV.
+                if len(rows[0]) != len(header):
+                    raise ValueError(
+                        f"CSV column count mismatch. Expected {len(header)} columns, "
+                        f"got {len(rows[0])} in input '{_name}'. "
+                        "All CSV inputs must have the same columns."
+                    )
+                all_rows.extend(rows[1:])
+
+        if not all_rows:
+            raise ValueError("No valid CSV data found in any input.")
+
+        output = io.StringIO()
+        writer = csv.writer(output)
+        writer.writerows(all_rows)
+        result = output.getvalue()
+
+    else:
+        raise ValueError(
+            f"Unknown output_type: '{output_type}'. Must be 'JsonArray', 'JsonObject', or 'CSV'."
+        )
+
+    os.makedirs(os.path.dirname(output_path), exist_ok=True)
+    with open(output_path, "w") as f:
+        f.write(result)
+
+
+if __name__ == "__main__":
+    # Note: this component uses dynamic **input_values (agg_N inputs) which cannot be
+    # auto-generated by create_component_from_func. The component.yaml is a hand-authored
+    # template that is extended at runtime with additional agg_N inputs. See component.yaml
+    # for the full container specification.
+    pass

components/dataset_manipulation/Aggregate_inputs/component.yaml

@@ -0,0 +1,141 @@
+# Input Aggregator component specification.
+# This file is a template; at runtime the spec is extended with additional inputs
+# (agg_1, agg_2, ...) based on the number of values to aggregate.
+# The inputs listed below represent the default template state.
+name: Input Aggregator
+description: |
+  Aggregates multiple pipeline inputs into a single output.
+  Supports JsonArray, JsonObject, and CSV aggregation modes.
+  Pass values as agg_N inputs and set output_type to control the aggregation format.
+metadata:
+  annotations:
+    is_input_aggregator: "true"
+    author: Tangle Team
+inputs:
+- {name: output_type, type: String, default: "JsonArray", optional: true, description: "Output format: JsonArray (array of values), JsonObject (keyed by input name), or CSV (union of CSV inputs with matching columns)."}
+# Example dynamically added input:
+# - {name: agg_1, type: String, optional: true}
+outputs:
+- {name: Output, type: String}
+implementation:
+  container:
+    image: python:3.12-slim
+    command:
+    - sh
+    - -ec
+    - |
+      program_path=$(mktemp)
+      printf "%s" "$0" > "$program_path"
+      python3 -u "$program_path" "$@"
+    - |
+      import csv
+      import io
+      import json
+      import os
+      import re
+      import sys
+
+
+      def aggregate(output_path: str, output_type: str = "JsonArray", **input_values: str) -> None:
+          inputs = {}
+          for key, value in sorted(input_values.items()):
+              if value is not None:
+                  inputs[key] = value
+
+          if not inputs:
+              raise ValueError("No inputs provided. At least one input must be connected.")
+
+          if output_type == "JsonArray":
+              result = json.dumps(list(inputs.values()), indent=2)
+
+          elif output_type == "JsonObject":
+              result = json.dumps(inputs, indent=2)
+
+          elif output_type == "CSV":
+              all_rows = []
+              header = None
+
+              def _normalize_csv_rows(rows: list[list[str]]) -> list[list[str]]:
+                  """Strip leading empty column (pandas index) from all rows if present."""
+                  if rows and rows[0] and rows[0][0] == "":
+                      return [row[1:] if row else row for row in rows]
+                  return rows
+
+              for _name, csv_text in inputs.items():
+                  text = csv_text.strip()
+                  if not text:
+                      continue
+                  reader = csv.reader(io.StringIO(text))
+                  rows = _normalize_csv_rows(list(reader))
+                  if not rows or rows == [[""]]:
+                      continue
+                  if header is None:
+                      header = rows[0]
+                      all_rows.append(header)
+                      all_rows.extend(rows[1:])
+                  else:
+                      # Always treat the first row of subsequent inputs as a header and skip it.
+                      # Raise only if column count differs, which would produce malformed CSV.
+                      if len(rows[0]) != len(header):
+                          raise ValueError(
+                              f"CSV column count mismatch. Expected {len(header)} columns, "
+                              f"got {len(rows[0])} in input '{_name}'. "
+                              "All CSV inputs must have the same columns."
+                          )
+                      all_rows.extend(rows[1:])
+
+              if not all_rows:
+                  raise ValueError("No valid CSV data found in any input.")
+
+              output = io.StringIO()
+              writer = csv.writer(output)
+              writer.writerows(all_rows)
+              result = output.getvalue()
+
+          else:
+              raise ValueError(
+                  f"Unknown output_type: '{output_type}'. Must be 'JsonArray', 'JsonObject', or 'CSV'."
+              )
+
+          os.makedirs(os.path.dirname(output_path), exist_ok=True)
+          with open(output_path, "w") as f:
+              f.write(result)
+
+
+      _args = sys.argv[1:]
+      _output_path = None
+      _output_type = "JsonArray"
+      _input_values = {}
+
+      _i = 0
+      while _i < len(_args):
+          _arg = _args[_i]
+
+          if _arg == "----output-paths" and _i + 1 < len(_args):
+              _output_path = _args[_i + 1]
+              _i += 2
+          elif _arg == "--output-type" and _i + 1 < len(_args):
+              _output_type = _args[_i + 1]
+              _i += 2
+          else:
+              # Match any --agg-N flag dynamically
+              _match = re.match(r"^--agg-(\d+)$", _arg)
+              if _match and _i + 1 < len(_args):
+                  _n = _match.group(1)
+                  _input_values[f"agg_{_n}"] = _args[_i + 1]
+                  _i += 2
+              else:
+                  _i += 1
+
+      if not _output_path:
+          raise ValueError("Output path is required (----output-paths).")
+
+      aggregate(
+          output_path=_output_path,
+          output_type=_output_type,
+          **_input_values,
+      )
+    - --output-type
+    - {inputValue: output_type}
+    - '----output-paths'
+    - {outputPath: Output}