fix(structured): anonymize columns whose name is not an identifier

presidio-structured/presidio_structured/data/data_processors.py

@@ -120,12 +120,12 @@ def _process(
 
         for key, operator_callable in key_to_operator_mapping.items():
             self.logger.debug(f"Operating on column {key}")
-            for row in data.itertuples(index=True):
-                text_to_operate_on = getattr(row, key)
+            for index in data.index:
+                text_to_operate_on = data.at[index, key]
                 operated_text = self._operate_on_text(
                     text_to_operate_on, operator_callable
                 )
-                data.at[row.Index, key] = operated_text
+                data.at[index, key] = operated_text
         return data
 
 

presidio-structured/tests/data/test_data_transformers.py

@@ -1,5 +1,6 @@
 import pytest
 from pandas import DataFrame
+from presidio_structured.config import StructuredAnalysis
 from presidio_structured.data.data_processors import (
     DataProcessorBase,
     PandasDataProcessor,
@@ -24,6 +25,25 @@ def test_process(self, sample_df, operators, tabular_analysis):
             else:
                 assert all(result[key] == "DEFAULT_REPLACEMENT")
 
+    def test_process_column_name_not_an_identifier(self, operators):
+        # Column names holding PII are often not valid Python identifiers
+        # ("Full Name", "e-mail", ...). itertuples renames those to positional
+        # fields, so a name-based getattr lookup skips them and the PII is left
+        # in place.
+        processor = PandasDataProcessor()
+        df = DataFrame(
+            {
+                "Full Name": ["John Doe", "Jane Doe"],
+                "email": ["john@example.com", "jane@example.com"],
+            }
+        )
+        analysis = StructuredAnalysis(
+            entity_mapping={"Full Name": "PERSON", "email": "EMAIL_ADDRESS"}
+        )
+        result = processor.operate(df, analysis, operators)
+        assert all(result["Full Name"] == "PERSON_REPLACEMENT")
+        assert all(result["email"] == "DEFAULT_REPLACEMENT")
+
     def test_process_no_default_should_raise(self, sample_df, operators_no_default, tabular_analysis):
         processor = PandasDataProcessor()
         with pytest.raises(ValueError):