def redact_pii_text()

in function_app/bp_pii_redaction.py [0:0]

• 53 lines of code
• 11 McCabe index (conditional complexity)

def redact_pii_text(req: func.HttpRequest) -> func.HttpResponse:
    logging.info(
        f"Python HTTP trigger function `{TEXT_FUNCTION_ROUTE}` received a request."
    )
    # Create the object to hold all intermediate and final values. We will progressively update
    # values as each stage of the pipeline is completed, allowing us to return a partial
    # response in case of an error at any stage.
    output_model = TextFunctionReponseModel(success=False)
    try:
        # Create error_text and error_code variables. These will be updated as
        # we move through the pipeline so that if a step fails, the vars reflect
        # what has failed. If all steps complete successfully, the vars are
        # never used.
        error_text = "An error occurred during processing."
        error_code = 422

        func_timer = MeasureRunTime()
        func_timer.start()

        ### 1. Check the request body
        req_body = req.get_json()
        try:
            request_obj = TextFunctionRequestModel(**req_body)
        except Exception as _e:
            raise ValueError(
                (
                    "The request body was not in the expected format. Please ensure that it is "
                    "a valid JSON object with the following fields: {}"
                ).format(list(TextFunctionRequestModel.model_fields.keys()))
            )

        ### 2. Create the messages to send to the LLM
        error_text = "An error occurred during PII recognition"
        error_code = 500

        documents = [request_obj.text]
        ### 3. Redact PII from the text using Azure AI Language service
        pii_result = text_analytics_client.recognize_pii_entities(
            documents=documents,
        )
        output_model.pii_raw_response = [str(doc_result) for doc_result in pii_result]
        pii_result_doc = pii_result[0]
        if pii_result_doc.is_error:
            raise RuntimeError(
                "An error occurred during PII recognition",
            )

        ### 4. Replace the PII entities with '<<CATEGORY>>' text.
        # This gives us a more readable output than the redacted text from the raw API response (which simply replaces PII with asterixes).
        replacements = {
            entity.text: f"<<{entity.category}>>" for entity in pii_result_doc.entities
        }
        output_model.redacted_text = replace_text(request_obj.text, replacements)

        ### 5. All steps completed successfully, set success=True and return the final result
        output_model.success = True
        output_model.func_time_taken_secs = func_timer.stop()

        return func.HttpResponse(
            body=output_model.model_dump_json(),
            mimetype="application/json",
            status_code=200,
        )
    except Exception as _e:
        # If an error occurred at any stage, return the partial response. Update the error_text
        # field to contain the error message, and ensure success=False.
        output_model.success = False
        output_model.error_text = error_text
        output_model.func_time_taken_secs = func_timer.stop()
        logging.exception(output_model.error_text)
        return func.HttpResponse(
            body=output_model.model_dump_json(),
            mimetype="application/json",
            status_code=error_code,
        )