Coverage for presidio_analyzer / predefined_recognizers / third_party / azure_openai_langextract_recognizer.py: 100%

1"""Azure OpenAI LangExtract Recognizer for PII/PHI detection.""" 

2 

3import logging 

4import os 

5from pathlib import Path 

6from typing import Optional 

7 

8try: 

9 import langextract as lx 

10 LANGEXTRACT_AVAILABLE = True 

11except ImportError: # pragma: no cover 

12 LANGEXTRACT_AVAILABLE = False 

13 lx = None 

14 

15from presidio_analyzer.predefined_recognizers.third_party import ( 

16 langextract_recognizer, 

17) 

18 

19# Import provider module to trigger registration with LangExtract 

20# This must happen at module import time for the @register decorator to execute 

21try: 

22 from presidio_analyzer.predefined_recognizers.third_party import ( 

23 azure_openai_provider, # noqa: F401 - imported for side effects (registration) 

24 ) 

25except ImportError: # pragma: no cover 

26 pass 

27 

28logger = logging.getLogger("presidio-analyzer") 

29 

30LangExtractRecognizer = langextract_recognizer.LangExtractRecognizer 

31 

32AZURE_OPENAI_DOCS_URL = "https://learn.microsoft.com/en-us/azure/ai-services/openai/" 

33 

34 

35class AzureOpenAILangExtractRecognizer(LangExtractRecognizer): 

36 """ 

37 Concrete implementation of LangExtract recognizer using Azure OpenAI backend. 

38 

39 Provides Azure OpenAI-specific functionality including: 

40 - Azure OpenAI endpoint and API key configuration 

41 - Deployment name management 

42 - API version control 

43 - Integration with custom Azure OpenAI provider via LangExtract registry 

44 """ 

45 

46 DEFAULT_API_VERSION = "2024-02-15-preview" 

47 

48 DEFAULT_CONFIG_PATH = ( 

49 Path(__file__).parent.parent.parent 

50 / "conf" 

51 / "langextract_config_azureopenai.yaml" 

52 ) 

53 

54 def __init__( 

55 self, 

56 model_id: Optional[str] = None, 

57 config_path: Optional[str] = None, 

58 azure_endpoint: Optional[str] = None, 

59 api_key: Optional[str] = None, 

60 api_version: Optional[str] = None, 

61 supported_language: str = "en", 

62 name: str = "Azure OpenAI LangExtract PII", 

63 ): 

64 """ 

65 Initialize Azure OpenAI LangExtract recognizer for PII/PHI detection. 

66 

67 Note: Azure OpenAI endpoint and API configuration are not validated 

68 during initialization. Any connectivity or authentication issues will 

69 be reported when analyze() is first called. 

70 

71 Configuration priority for authentication (highest to lowest): 

72 1. Direct parameters (azure_endpoint, api_key, api_version) 

73 2. Environment variables (AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_API_KEY, 

74 AZURE_OPENAI_API_VERSION) 

75 

76 Configuration priority for model_id (deployment name): 

77 1. Direct parameter (model_id) 

78 2. Config file (langextract.model.model_id) 

79 

80 :param model_id: Azure OpenAI deployment name (e.g., "gpt-4", 

81 "gpt-4o"). Overrides config file. 

82 :param config_path: Path to YAML configuration file. If not provided, 

83 uses default config. 

84 :param azure_endpoint: Azure OpenAI endpoint URL (overrides env var). 

85 :param api_key: Azure OpenAI API key (overrides env var). If not 

86 provided, uses managed identity. 

87 :param api_version: Azure OpenAI API version (optional, defaults to 

88 "2024-02-15-preview"). 

89 :param supported_language: Language this recognizer supports 

90 (optional, default: "en"). 

91 :raises ImportError: If langextract is not installed. 

92 :raises ValueError: If Azure OpenAI endpoint is not provided via 

93 parameter or env var. 

94 """ 

95 

96 # Determine actual config path 

97 actual_config_path = ( 

98 config_path if config_path else str(self.DEFAULT_CONFIG_PATH) 

99 ) 

100 

101 # Get Azure-specific settings with priority: parameter > env var 

102 self.azure_endpoint = azure_endpoint or os.environ.get("AZURE_OPENAI_ENDPOINT") 

103 self._validate_azure_endpoint(self.azure_endpoint) 

104 

105 self.api_key = api_key or os.environ.get("AZURE_OPENAI_API_KEY") 

106 self.api_version = ( 

107 api_version 

108 or os.environ.get("AZURE_OPENAI_API_VERSION") 

109 or self.DEFAULT_API_VERSION 

110 ) 

111 

112 # Initialize parent class (loads config, sets self.model_id from config) 

113 super().__init__( 

114 config_path=actual_config_path, 

115 name=name, 

116 supported_language=supported_language, 

117 extract_params={ 

118 "extract": { 

119 "fence_output": True, 

120 "use_schema_constraints": False, 

121 }, 

122 }, 

123 ) 

124 

125 # Override model_id if provided as parameter (deployment name) 

126 if model_id: 

127 self.model_id = model_id 

128 

129 def _validate_azure_endpoint(self, azure_endpoint: Optional[str]) -> None: 

130 """ 

131 Validate that Azure OpenAI endpoint is provided. 

132 

133 :param azure_endpoint: Azure OpenAI endpoint URL. 

134 :raises ValueError: If endpoint is not provided. 

135 """ 

136 if not azure_endpoint: 

137 raise ValueError( 

138 "Azure OpenAI endpoint is required. Provide 'azure_endpoint' parameter " 

139 f"or set AZURE_OPENAI_ENDPOINT environment variable.\n" 

140 f"See {AZURE_OPENAI_DOCS_URL} for details." 

141 ) 

142 

143 def _get_provider_params(self): 

144 """Return Azure OpenAI-specific params.""" 

145 model_id_with_prefix = f"azure:{self.model_id}" 

146 

147 language_model_params = { 

148 "azure_endpoint": self.azure_endpoint, 

149 "api_version": self.api_version, 

150 "azure_deployment": self.model_id, 

151 } 

152 

153 if self.api_key: 

154 language_model_params["api_key"] = self.api_key 

155 

156 return { 

157 "model_id": model_id_with_prefix, 

158 "language_model_params": language_model_params, 

159 }