Coverage for src/qdrant_loader/core/chunking/strategy/code/metadata/performance.py: 90%

1from __future__ import annotations 

2 

3import re 

4 

5 

6def analyze_performance_patterns(content: str) -> dict[str, list[str]]: 

7 performance_indicators = { 

8 "optimization_patterns": [], 

9 "potential_bottlenecks": [], 

10 "resource_usage": [], 

11 } 

12 

13 content_lower = content.lower() 

14 

15 if any(k in content_lower for k in ["cache", "memoize", "lazy"]): 

16 performance_indicators["optimization_patterns"].append("caching") 

17 if "async" in content_lower or "await" in content_lower: 

18 performance_indicators["optimization_patterns"].append("async_programming") 

19 if any(k in content_lower for k in ["parallel", "concurrent", "threading"]): 

20 performance_indicators["optimization_patterns"].append("concurrency") 

21 

22 total_loops = content.count("for ") + content.count("while ") 

23 if total_loops >= 3: 

24 performance_indicators["potential_bottlenecks"].append("nested_loops") 

25 

26 lines = content.split("\n") 

27 def_pattern = re.compile(r"^\s*(?:async\s+)?def\s+([a-zA-Z_][a-zA-Z0-9_]*)\s*\(") 

28 for idx, line in enumerate(lines): 

29 match = def_pattern.match(line) 

30 if not match: 

31 continue 

32 func_name = match.group(1) 

33 bare_call_regex = re.compile(r"\b" + re.escape(func_name) + r"\s*\(") 

34 method_call_regex = re.compile(r"\." + re.escape(func_name) + r"\s*\(") 

35 call_count = 0 

36 for j, other_line in enumerate(lines): 

37 if j == idx: 

38 continue 

39 if bare_call_regex.search(other_line) or method_call_regex.search( 

40 other_line 

41 ): 

42 call_count += 1 

43 if call_count > 0: 

44 performance_indicators["potential_bottlenecks"].append("recursion") 

45 break 

46 

47 if content.count("database") > 5 or content.count("query") > 5: 

48 performance_indicators["potential_bottlenecks"].append("database_heavy") 

49 if content.count("file") > 10 or content.count("read") > 10: 

50 performance_indicators["potential_bottlenecks"].append("io_heavy") 

51 

52 if any(k in content_lower for k in ["memory", "buffer", "allocation"]): 

53 performance_indicators["resource_usage"].append("memory_allocation") 

54 if any(k in content_lower for k in ["connection", "pool", "socket"]): 

55 performance_indicators["resource_usage"].append("connection_management") 

56 

57 return performance_indicators