Coverage for src/qdrant_loader/connectors/confluence/pagination.py: 74%

1from __future__ import annotations 

2 

3import re 

4from typing import Any 

5 

6_ALLOWED_TOKEN_RE = re.compile(r"^[A-Za-z0-9_-]+$") 

7 

8# Full expand used when the page body/content is needed (streaming, fetch_by_id). 

9CONTENT_EXPAND = ( 

10 "body.storage,version,metadata.labels,history,space,extensions.position," 

11 "children.comment.body.storage,ancestors,children.page" 

12) 

13 

14# Minimal expand used when only enough metadata to filter by labels is needed 

15# (e.g. list_entity_ids), avoiding fetching full page bodies. 

16LIGHT_EXPAND = "metadata.labels" 

17 

18 

19def _quote_cql_literal(value: str) -> str: 

20 # Escape backslashes first, then double quotes 

21 escaped = value.replace("\\", "\\\\").replace('"', '\\"') 

22 return f'"{escaped}"' 

23 

24 

25def _sanitize_space_key(space_key: str) -> str: 

26 if not _ALLOWED_TOKEN_RE.fullmatch(space_key): 

27 raise ValueError( 

28 "Invalid Confluence space key. Only alphanumerics, underscore and hyphen are allowed." 

29 ) 

30 return _quote_cql_literal(space_key) 

31 

32 

33def _sanitize_content_types(content_types: list[str]) -> list[str]: 

34 sanitized: list[str] = [] 

35 for content_type in content_types: 

36 if not isinstance(content_type, str) or not _ALLOWED_TOKEN_RE.fullmatch( 

37 content_type 

38 ): 

39 raise ValueError(f"Invalid Confluence content type: {content_type!r}") 

40 sanitized.append(_quote_cql_literal(content_type)) 

41 return sanitized 

42 

43 

44def build_cloud_search_params( 

45 space_key: str, 

46 content_types: list[str] | None, 

47 cursor: str | None, 

48 light: bool = False, 

49) -> dict[str, Any]: 

50 params: dict[str, Any] = { 

51 "expand": LIGHT_EXPAND if light else CONTENT_EXPAND, 

52 "limit": 25, 

53 } 

54 cql = f"space = {_sanitize_space_key(space_key)}" 

55 if content_types: 

56 safe_types = _sanitize_content_types(content_types) 

57 cql += f" and type in ({','.join(safe_types)})" 

58 params["cql"] = cql 

59 if cursor is not None: 

60 params["cursor"] = cursor 

61 return params 

62 

63 

64def build_dc_search_params( 

65 space_key: str, 

66 content_types: list[str] | None, 

67 start: int, 

68 light: bool = False, 

69) -> dict[str, Any]: 

70 params: dict[str, Any] = { 

71 "expand": LIGHT_EXPAND if light else CONTENT_EXPAND, 

72 "limit": 25, 

73 "start": start, 

74 } 

75 cql = f"space = {_sanitize_space_key(space_key)}" 

76 if content_types: 

77 safe_types = _sanitize_content_types(content_types) 

78 cql += f" and type in ({','.join(safe_types)})" 

79 params["cql"] = cql 

80 return params