Presidio Anonymizer API Reference

Anonymizer root module.

`anonymizer_engine`

Handles the entire logic of the Presidio-anonymizer and text anonymizing.

`AnonymizerEngine (EngineBase)`

AnonymizerEngine class.

Handles the entire logic of the Presidio-anonymizer. Gets the original text and replaces the PII entities with the desired anonymizers.

Source code in presidio_anonymizer/anonymizer_engine.py

class AnonymizerEngine(EngineBase):
    """
    AnonymizerEngine class.

    Handles the entire logic of the Presidio-anonymizer. Gets the original text
    and replaces the PII entities with the desired anonymizers.
    """

    logger = logging.getLogger("presidio-anonymizer")

    def __init__(self):
        EngineBase.__init__(self)

    def anonymize(
        self,
        text: str,
        analyzer_results: List[RecognizerResult],
        operators: Optional[Dict[str, OperatorConfig]] = None,
    ) -> EngineResult:
        """Anonymize method to anonymize the given text.

        :param text: the text we are anonymizing
        :param analyzer_results: A list of RecognizerResult class -> The results we
        received from the analyzer
        :param operators: The configuration of the anonymizers we would like
        to use for each entity e.g.: {"PHONE_NUMBER":OperatorConfig("redact", {})}
        received from the analyzer
        :return: the anonymized text and a list of information about the
        anonymized entities.

        :example:

        >>> from presidio_anonymizer import AnonymizerEngine
        >>> from presidio_anonymizer.entities import RecognizerResult, OperatorConfig

        >>> # Initialize the engine with logger.
        >>> engine = AnonymizerEngine()

        >>> # Invoke the anonymize function with the text, analyzer results and
        >>> # Operators to define the anonymization type.
        >>> result = engine.anonymize(
        >>>     text="My name is Bond, James Bond",
        >>>     analyzer_results=[RecognizerResult(entity_type="PERSON",
        >>>                                        start=11,
        >>>                                        end=15,
        >>>                                        score=0.8),
        >>>                       RecognizerResult(entity_type="PERSON",
        >>>                                        start=17,
        >>>                                        end=27,
        >>>                                        score=0.8)],
        >>>     operators={"PERSON": OperatorConfig("replace", {"new_value": "BIP"})}
        >>> )

        >>> print(result)
        text: My name is BIP, BIP.
        items:
        [
            {'start': 16, 'end': 19, 'entity_type': 'PERSON',
             'text': 'BIP', 'operator': 'replace'},
            {'start': 11, 'end': 14, 'entity_type': 'PERSON',
             'text': 'BIP', 'operator': 'replace'}
        ]


        """
        analyzer_results = self._remove_conflicts_and_get_text_manipulation_data(
            analyzer_results
        )

        operators = self.__check_or_add_default_operator(operators)

        return self._operate(text, analyzer_results, operators, OperatorType.Anonymize)

    def _remove_conflicts_and_get_text_manipulation_data(
        self, analyzer_results: List[RecognizerResult]
    ) -> List[RecognizerResult]:
        """
        Iterate the list and create a sorted unique results list from it.

        Only insert results which are:
        1. Indices are not contained in other result.
        2. Have the same indices as other results but with larger score.
        :return: List
        """
        tmp_analyzer_results = []
        # This list contains all elements which we need to check a single result
        # against. If a result is dropped, it can also be dropped from this list
        # since it is intersecting with another result and we selected the other one.
        other_elements = analyzer_results.copy()
        for result in analyzer_results:
            other_elements.remove(result)

            is_merge_same_entity_type = False
            for other_element in other_elements:
                if other_element.entity_type != result.entity_type:
                    continue
                if result.intersects(other_element) == 0:
                    continue

                other_element.start = min(result.start, other_element.start)
                other_element.end = max(result.end, other_element.end)
                other_element.score = max(result.score, other_element.score)
                is_merge_same_entity_type = True
                break
            if not is_merge_same_entity_type:
                other_elements.append(result)
                tmp_analyzer_results.append(result)
            else:
                self.logger.debug(f"removing element {result} from "
                                  f"results list due to merge")

        unique_text_metadata_elements = []
        # This list contains all elements which we need to check a single result
        # against. If a result is dropped, it can also be dropped from this list
        # since it is intersecting with another result and we selected the other one.
        other_elements = tmp_analyzer_results.copy()
        for result in tmp_analyzer_results:
            other_elements.remove(result)
            result_conflicted = self.__is_result_conflicted_with_other_elements(
                other_elements, result
            )
            if not result_conflicted:
                other_elements.append(result)
                unique_text_metadata_elements.append(result)
            else:
                self.logger.debug(
                    f"removing element {result} from results list due to conflict"
                )
        return unique_text_metadata_elements

    def get_anonymizers(self) -> List[str]:
        """Return a list of supported anonymizers."""
        names = [p for p in self.operators_factory.get_anonymizers().keys()]
        return names

    @staticmethod
    def __is_result_conflicted_with_other_elements(other_elements, result):
        return any(
            [result.has_conflict(other_element) for other_element in other_elements]
        )

    @staticmethod
    def __check_or_add_default_operator(
        operators: Dict[str, OperatorConfig]
    ) -> Dict[str, OperatorConfig]:
        default_operator = OperatorConfig(DEFAULT)
        if not operators:
            return {"DEFAULT": default_operator}
        if not operators.get("DEFAULT"):
            operators["DEFAULT"] = default_operator
        return operators

`anonymize(self, text, analyzer_results, operators=None)`

Anonymize method to anonymize the given text.

:param text: the text we are anonymizing :param analyzer_results: A list of RecognizerResult class -> The results we received from the analyzer :param operators: The configuration of the anonymizers we would like to use for each entity e.g.: {"PHONE_NUMBER":OperatorConfig("redact", {})} received from the analyzer :return: the anonymized text and a list of information about the anonymized entities.

:example:

from presidio_anonymizer import AnonymizerEngine from presidio_anonymizer.entities import RecognizerResult, OperatorConfig

Initialize the engine with logger.

engine = AnonymizerEngine()

Invoke the anonymize function with the text, analyzer results and

Operators to define the anonymization type.

result = engine.anonymize( text="My name is Bond, James Bond", analyzer_results=[RecognizerResult(entity_type="PERSON", start=11, end=15, score=0.8), RecognizerResult(entity_type="PERSON", start=17, end=27, score=0.8)], operators={"PERSON": OperatorConfig("replace", {"new_value": "BIP"})} )

print(result) text: My name is BIP, BIP. items: [ {'start': 16, 'end': 19, 'entity_type': 'PERSON', 'text': 'BIP', 'operator': 'replace'}, {'start': 11, 'end': 14, 'entity_type': 'PERSON', 'text': 'BIP', 'operator': 'replace'} ]

Source code in presidio_anonymizer/anonymizer_engine.py

def anonymize(
    self,
    text: str,
    analyzer_results: List[RecognizerResult],
    operators: Optional[Dict[str, OperatorConfig]] = None,
) -> EngineResult:
    """Anonymize method to anonymize the given text.

    :param text: the text we are anonymizing
    :param analyzer_results: A list of RecognizerResult class -> The results we
    received from the analyzer
    :param operators: The configuration of the anonymizers we would like
    to use for each entity e.g.: {"PHONE_NUMBER":OperatorConfig("redact", {})}
    received from the analyzer
    :return: the anonymized text and a list of information about the
    anonymized entities.

    :example:

    >>> from presidio_anonymizer import AnonymizerEngine
    >>> from presidio_anonymizer.entities import RecognizerResult, OperatorConfig

    >>> # Initialize the engine with logger.
    >>> engine = AnonymizerEngine()

    >>> # Invoke the anonymize function with the text, analyzer results and
    >>> # Operators to define the anonymization type.
    >>> result = engine.anonymize(
    >>>     text="My name is Bond, James Bond",
    >>>     analyzer_results=[RecognizerResult(entity_type="PERSON",
    >>>                                        start=11,
    >>>                                        end=15,
    >>>                                        score=0.8),
    >>>                       RecognizerResult(entity_type="PERSON",
    >>>                                        start=17,
    >>>                                        end=27,
    >>>                                        score=0.8)],
    >>>     operators={"PERSON": OperatorConfig("replace", {"new_value": "BIP"})}
    >>> )

    >>> print(result)
    text: My name is BIP, BIP.
    items:
    [
        {'start': 16, 'end': 19, 'entity_type': 'PERSON',
         'text': 'BIP', 'operator': 'replace'},
        {'start': 11, 'end': 14, 'entity_type': 'PERSON',
         'text': 'BIP', 'operator': 'replace'}
    ]


    """
    analyzer_results = self._remove_conflicts_and_get_text_manipulation_data(
        analyzer_results
    )

    operators = self.__check_or_add_default_operator(operators)

    return self._operate(text, analyzer_results, operators, OperatorType.Anonymize)

`get_anonymizers(self)`

Return a list of supported anonymizers.

Source code in presidio_anonymizer/anonymizer_engine.py

def get_anonymizers(self) -> List[str]:
    """Return a list of supported anonymizers."""
    names = [p for p in self.operators_factory.get_anonymizers().keys()]
    return names

`core` `special`

The core text functionality.

`engine_base`

Handle the entire text operations using the operators.

`EngineBase (ABC)`

Handle the logic of operations over the text using the operators.

Source code in presidio_anonymizer/core/engine_base.py

class EngineBase(ABC):
    """Handle the logic of operations over the text using the operators."""

    def __init__(self):
        self.logger = logging.getLogger("presidio-anonymizer")
        self.operators_factory = OperatorsFactory()

    def _operate(
        self,
        text: str,
        pii_entities: List[PIIEntity],
        operators_metadata: Dict[str, OperatorConfig],
        operator_type: OperatorType,
    ) -> EngineResult:
        """
        Operate will do the operations required by the user over the text.

        :param text: the text we need to operate on.
        :param pii_entities: data about the text entities we want to operate over.
        :param operators_metadata: dictionary where the key is the entity_type and what
        :type operator_type: either anonymize or deanonymize
        we want to perform over this entity_type.
        :return:
        """
        text_replace_builder = TextReplaceBuilder(original_text=text)
        engine_result = EngineResult()
        sorted_pii_entities = sorted(pii_entities, reverse=True)
        for operator in sorted_pii_entities:
            text_to_operate_on = text_replace_builder.get_text_in_position(
                operator.start, operator.end
            )

            self.logger.debug(f"performing operation {operator}")
            operator_metadata = self.__get_entity_operator_metadata(
                operator.entity_type, operators_metadata
            )
            changed_text = self.__operate_on_text(
                operator, text_to_operate_on, operator_metadata, operator_type
            )
            index_from_end = text_replace_builder.replace_text_get_insertion_index(
                changed_text, operator.start, operator.end
            )

            # The following creates an intermediate list of result entities,
            # ordered from end to start, and the indexes will be normalized
            # from start to end once the loop ends and the text length is deterministic.
            result_item = OperatorResult(
                0,
                index_from_end,
                operator.entity_type,
                changed_text,
                operator_metadata.operator_name,
            )
            engine_result.add_item(result_item)

        engine_result.set_text(text_replace_builder.output_text)
        engine_result.normalize_item_indexes()
        return engine_result

    def __operate_on_text(
        self,
        text_metadata: PIIEntity,
        text_to_operate_on: str,
        operator_metadata: OperatorConfig,
        operator_type: OperatorType,
    ) -> str:
        entity_type = text_metadata.entity_type
        self.logger.debug(f"getting operator for {entity_type}")
        operator = self.operators_factory.create_operator_class(
            operator_metadata.operator_name, operator_type
        )
        self.logger.debug(f"validating operator {operator} for {entity_type}")
        operator.validate(params=operator_metadata.params)
        params = operator_metadata.params
        params["entity_type"] = entity_type
        self.logger.debug(f"operating on {entity_type} with {operator}")
        operated_on_text = operator.operate(params=params, text=text_to_operate_on)
        return operated_on_text

    @staticmethod
    def __get_entity_operator_metadata(
        entity_type: str, operators_metadata: Dict = None
    ) -> OperatorConfig:
        # We try to get the operator from the list by entity_type.
        # If it does not exist, we get the default from the list.
        if operators_metadata is None:
            operators_metadata = {}
        operator = operators_metadata.get(entity_type)
        if operator:
            return operator
        else:
            return operators_metadata.get("DEFAULT")

`text_replace_builder`

Handles the original text and creates a new one according to changes requests.

`TextReplaceBuilder`

Creates new text according to users request.

Source code in presidio_anonymizer/core/text_replace_builder.py

class TextReplaceBuilder:
    """Creates new text according to users request."""

    def __init__(self, original_text: str):
        self.logger = logging.getLogger("presidio-anonymizer")
        self.output_text = original_text
        self.original_text = original_text
        self.text_len = len(original_text)
        self.last_replacement_index = self.text_len

    def get_text_in_position(self, start: int, end: int) -> str:
        """
        Get part of the text inside the original text.

        :param start: start position of inner text
        :param end: end position of inner text
        :return: str - part of the original text
        """
        self.__validate_position_in_text(start, end)
        return self.original_text[start:end]

    def replace_text_get_insertion_index(
        self, replacement_text: str, start: int, end: int
    ) -> int:
        """
        Replace text in a specific position with the text.

        :param replacement_text: new text to replace the old text according to indices
        :param start: the startpoint to replace the text
        :param end: the endpoint to replace the text
        :return: The index of inserted text
        """
        end_of_text_index = min(end, self.last_replacement_index)
        self.last_replacement_index = start

        before_text = self.output_text[:start]
        after_text = self.output_text[end_of_text_index:]
        self.output_text = before_text + replacement_text + after_text

        # The replace algorithm is replacing the text from end to start.
        # calculate and return the start point from the end.
        return len(after_text) + len(replacement_text)

    def __validate_position_in_text(self, start: int, end: int):
        """Validate the start and end position match the text length."""
        if self.text_len < start or end > self.text_len:
            err_msg = (
                f"Invalid analyzer result, start: {start} and end: "
                f"{end}, while text length is only {self.text_len}."
            )
            raise InvalidParamException(err_msg)

`get_text_in_position(self, start, end)`

Get part of the text inside the original text.

:param start: start position of inner text :param end: end position of inner text :return: str - part of the original text

Source code in presidio_anonymizer/core/text_replace_builder.py

def get_text_in_position(self, start: int, end: int) -> str:
    """
    Get part of the text inside the original text.

    :param start: start position of inner text
    :param end: end position of inner text
    :return: str - part of the original text
    """
    self.__validate_position_in_text(start, end)
    return self.original_text[start:end]

`replace_text_get_insertion_index(self, replacement_text, start, end)`

Replace text in a specific position with the text.

:param replacement_text: new text to replace the old text according to indices :param start: the startpoint to replace the text :param end: the endpoint to replace the text :return: The index of inserted text

Source code in presidio_anonymizer/core/text_replace_builder.py

def replace_text_get_insertion_index(
    self, replacement_text: str, start: int, end: int
) -> int:
    """
    Replace text in a specific position with the text.

    :param replacement_text: new text to replace the old text according to indices
    :param start: the startpoint to replace the text
    :param end: the endpoint to replace the text
    :return: The index of inserted text
    """
    end_of_text_index = min(end, self.last_replacement_index)
    self.last_replacement_index = start

    before_text = self.output_text[:start]
    after_text = self.output_text[end_of_text_index:]
    self.output_text = before_text + replacement_text + after_text

    # The replace algorithm is replacing the text from end to start.
    # calculate and return the start point from the end.
    return len(after_text) + len(replacement_text)

`deanonymize_engine`

Deanonymize anonymized text by using deanonymize operators.

`DeanonymizeEngine (EngineBase)`

Deanonymize text that was previously anonymized.

Source code in presidio_anonymizer/deanonymize_engine.py

class DeanonymizeEngine(EngineBase):
    """Deanonymize text that was previously anonymized."""

    def __init__(self):
        self.logger = logging.getLogger("presidio-anonymizer")
        EngineBase.__init__(self)

    def deanonymize(
        self,
        text: str,
        entities: List[OperatorResult],
        operators: Dict[str, OperatorConfig],
    ) -> EngineResult:
        """
        Receive the text, entities and operators to perform deanonymization over.

        :param operators: the operators to apply on the anonymizer result entities
        :param text: the full text with the encrypted entities
        :param entities: list of encrypted entities
        :return: EngineResult - the new text and data about the deanonymized entities.
        """
        return self._operate(text, entities, operators, OperatorType.Deanonymize)

    def get_deanonymizers(self) -> List[str]:
        """Return a list of supported deanonymizers."""
        names = [p for p in self.operators_factory.get_deanonymizers().keys()]
        return names

`deanonymize(self, text, entities, operators)`

Receive the text, entities and operators to perform deanonymization over.

:param operators: the operators to apply on the anonymizer result entities :param text: the full text with the encrypted entities :param entities: list of encrypted entities :return: EngineResult - the new text and data about the deanonymized entities.

Source code in presidio_anonymizer/deanonymize_engine.py

def deanonymize(
    self,
    text: str,
    entities: List[OperatorResult],
    operators: Dict[str, OperatorConfig],
) -> EngineResult:
    """
    Receive the text, entities and operators to perform deanonymization over.

    :param operators: the operators to apply on the anonymizer result entities
    :param text: the full text with the encrypted entities
    :param entities: list of encrypted entities
    :return: EngineResult - the new text and data about the deanonymized entities.
    """
    return self._operate(text, entities, operators, OperatorType.Deanonymize)

`get_deanonymizers(self)`

Return a list of supported deanonymizers.

Source code in presidio_anonymizer/deanonymize_engine.py

def get_deanonymizers(self) -> List[str]:
    """Return a list of supported deanonymizers."""
    names = [p for p in self.operators_factory.get_deanonymizers().keys()]
    return names

`entities` `special`

Handles all the entities objects (structs) of the anonymizer.

`engine` `special`

Engine request entities.

`operator_config`

`OperatorConfig`

Hold the data of the required operator.

Source code in presidio_anonymizer/entities/engine/operator_config.py

class OperatorConfig:
    """Hold the data of the required operator."""

    def __init__(self, operator_name: str, params: Dict = None):
        """
        Create an operator config instance.

        :param operator_name: the name of the operator we want to work with
        :param params: the parameters the operator needs in order to work
        """
        self.logger = logging.getLogger("presidio-anonymizer")
        self.operator_name = operator_name
        if not params:
            params = {}
        self.params = params
        self.__validate_fields()

    def __repr__(self):
        """Return a string representation of the object."""
        return f"operator_name: {self.operator_name}, params: {self.params}"

    @classmethod
    def from_json(cls, params: Dict) -> "OperatorConfig":
        """
        Create OperatorConfig from json.

        :param params: json e.g.: {
            "type": "mask",
            "masking_char": "*",
            "chars_to_mask": 4,
            "from_end": true
        }
        :return: OperatorConfig
        """
        operator_name = params.get("type")
        if operator_name:
            params.pop("type")
        return cls(operator_name, params)

    def __eq__(self, other: "OperatorConfig"):
        """Verify two OperatorConfigs are equal."""
        operator_name = self.operator_name == other.operator_name
        return self.params == other.params and operator_name

    def __validate_fields(self):
        validate_parameter_not_empty(
            self.operator_name, "operator config", "operator_name"
        )

`eq(self, other)` `special`

Verify two OperatorConfigs are equal.

Source code in presidio_anonymizer/entities/engine/operator_config.py

def __eq__(self, other: "OperatorConfig"):
    """Verify two OperatorConfigs are equal."""
    operator_name = self.operator_name == other.operator_name
    return self.params == other.params and operator_name

`init(self, operator_name, params=None)` `special`

Create an operator config instance.

:param operator_name: the name of the operator we want to work with :param params: the parameters the operator needs in order to work

Source code in presidio_anonymizer/entities/engine/operator_config.py

def __init__(self, operator_name: str, params: Dict = None):
    """
    Create an operator config instance.

    :param operator_name: the name of the operator we want to work with
    :param params: the parameters the operator needs in order to work
    """
    self.logger = logging.getLogger("presidio-anonymizer")
    self.operator_name = operator_name
    if not params:
        params = {}
    self.params = params
    self.__validate_fields()

`repr(self)` `special`

Return a string representation of the object.

Source code in presidio_anonymizer/entities/engine/operator_config.py

def __repr__(self):
    """Return a string representation of the object."""
    return f"operator_name: {self.operator_name}, params: {self.params}"

`from_json(params)` `classmethod`

Create OperatorConfig from json.

:param params: json e.g.: { "type": "mask", "masking_char": "*", "chars_to_mask": 4, "from_end": true } :return: OperatorConfig

Source code in presidio_anonymizer/entities/engine/operator_config.py

@classmethod
def from_json(cls, params: Dict) -> "OperatorConfig":
    """
    Create OperatorConfig from json.

    :param params: json e.g.: {
        "type": "mask",
        "masking_char": "*",
        "chars_to_mask": 4,
        "from_end": true
    }
    :return: OperatorConfig
    """
    operator_name = params.get("type")
    if operator_name:
        params.pop("type")
    return cls(operator_name, params)

`pii_entity`

`PIIEntity (ABC)`

Abstract class to hold the text we are going to operate on metadata.

Source code in presidio_anonymizer/entities/engine/pii_entity.py

class PIIEntity(ABC):
    """Abstract class to hold the text we are going to operate on metadata."""

    logger = logging.getLogger("presidio-anonymizer")

    def __init__(self, start: int, end: int, entity_type: str):
        self.start = start
        self.end = end
        self.entity_type = entity_type
        self.__validate_fields()

    def __repr__(self):
        """Return a string representation of the object."""
        return (
            f"start: {self.start}"
            f"end: {self.end},"
            f"entity_type: {self.entity_type}"
        )

    def __gt__(self, other):
        """Check one entity is greater then other by the text end index."""
        return self.start > other.start

    def __eq__(self, other):
        """Check two text metadata entities are equal."""
        return (
            self.start == other.start
            and self.end == other.end
            and self.entity_type == other.entity_type
        )

    def __validate_fields(self):
        validate_parameter_exists(self.start, "result", "start")
        validate_type(self.start, "start", int)
        validate_parameter_exists(self.end, "result", "end")
        validate_type(self.end, "end", int)
        validate_parameter_not_empty(self.entity_type, "result", "entity_type")
        if self.start < 0 or self.end < 0:
            raise InvalidParamException(
                "Invalid input, result start and end must be positive"
            )
        if self.start > self.end:
            raise InvalidParamException(
                f"Invalid input, start index '{self.start}' "
                f"must be smaller than end index '{self.end}'"
            )

`eq(self, other)` `special`

Check two text metadata entities are equal.

Source code in presidio_anonymizer/entities/engine/pii_entity.py

def __eq__(self, other):
    """Check two text metadata entities are equal."""
    return (
        self.start == other.start
        and self.end == other.end
        and self.entity_type == other.entity_type
    )

`gt(self, other)` `special`

Check one entity is greater then other by the text end index.

Source code in presidio_anonymizer/entities/engine/pii_entity.py

def __gt__(self, other):
    """Check one entity is greater then other by the text end index."""
    return self.start > other.start

`repr(self)` `special`

Return a string representation of the object.

Source code in presidio_anonymizer/entities/engine/pii_entity.py

def __repr__(self):
    """Return a string representation of the object."""
    return (
        f"start: {self.start}"
        f"end: {self.end},"
        f"entity_type: {self.entity_type}"
    )

`recognizer_result`

RecognizerResult is an exact copy of the RecognizerResult object from presidio-analyzer.

Represents the findings of detected entity.

`RecognizerResult (PIIEntity)`

Recognizer Result represents the findings of the detected entity.

Result of a recognizer analyzing the text.

:param entity_type: the type of the entity :param start: the start location of the detected entity :param end: the end location of the detected entity :param score: the score of the detection