Coverage for /usr/local/lib/python3.12/site-packages/prefect/logging/filters.py: 33%

1import logging 1 ctx1a

2from typing import Any 1 ctx1a

3 

4from prefect.utilities.collections import visit_collection 1 ctx1a

5from prefect.utilities.names import obfuscate 1 ctx1a

6 

7 

8def redact_substr(obj: Any, substr: str) -> Any: 1 ctx1a

9 """ 

10 Redact a string from a potentially nested object. 

11 

12 Args: 

13 obj: The object to redact the string from 

14 substr: The string to redact. 

15 

16 Returns: 

17 Any: The object with the API key redacted. 

18 """ 

19 

20 def redact_item(item: Any) -> Any: 

21 if isinstance(item, str): 

22 return item.replace(substr, obfuscate(substr)) 

23 return item 

24 

25 redacted_obj = visit_collection(obj, visit_fn=redact_item, return_data=True) 

26 return redacted_obj 

27 

28 

29class ObfuscateApiKeyFilter(logging.Filter): 1 ctx1a

30 """ 

31 A logging filter that obfuscates any string that matches the obfuscate_string function. 

32 """ 

33 

34 def filter(self, record: logging.LogRecord) -> bool: 1 ctx1a

35 # Need to import here to avoid circular imports 

36 from prefect.settings import PREFECT_API_KEY 

37 

38 if PREFECT_API_KEY: 

39 record.msg = redact_substr(record.msg, PREFECT_API_KEY.value()) 

40 

41 return True