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

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

* fix(structured): iterate column Series instead of per-cell .at reads

---------

Co-authored-by: Ron Shakutai <58519179+RonShakutai@users.noreply.github.com>
Co-authored-by: Omri Mendels <omri374@users.noreply.github.com>
Co-authored-by: Sharon Hart <sharonh.dev@gmail.com>
This commit is contained in:
uwezkhan
2026-07-23 14:38:52 +05:30
committed by GitHub
parent 88ebae34d0
commit cc15735f99
2 changed files with 22 additions and 3 deletions

View File

@@ -120,12 +120,11 @@ class PandasDataProcessor(DataProcessorBase):
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, text_to_operate_on in data[key].items():
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

View File

@@ -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 @@ class TestPandasDataProcessor:
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 renamed those to positional
# fields, so the name-based getattr lookup raised AttributeError,
# aborting the run with any later PII columns left unredacted.
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):