Coverage for src/qdrant_loader/connectors/confluence/auth.py: 38%

1from __future__ import annotations 

2 

3from requests import Session 

4from requests.auth import HTTPBasicAuth 

5 

6from .config import ConfluenceDeploymentType, ConfluenceSpaceConfig 

7 

8 

9def setup_authentication(session: Session, config: ConfluenceSpaceConfig) -> None: 

10 """Configure authentication on the provided requests session based on deployment type.""" 

11 if config.deployment_type == ConfluenceDeploymentType.CLOUD: 

12 if not config.token: 

13 raise ValueError("API token is required for Confluence Cloud") 

14 if not config.email: 

15 raise ValueError("Email is required for Confluence Cloud") 

16 session.auth = HTTPBasicAuth(config.email, config.token) 

17 else: 

18 if not config.token: 

19 raise ValueError( 

20 "Personal Access Token is required for Confluence Data Center/Server" 

21 ) 

22 session.headers.update( 

23 { 

24 "Authorization": f"Bearer {config.token}", 

25 } 

26 ) 

27 

28 

29def auto_detect_deployment_type(base_url: str) -> ConfluenceDeploymentType: 

30 """Auto-detect the Confluence deployment type based on base URL.""" 

31 from urllib.parse import urlparse 

32 

33 try: 

34 parsed_url = urlparse(str(base_url)) 

35 hostname = parsed_url.hostname 

36 if hostname is None: 

37 return ConfluenceDeploymentType.DATACENTER 

38 if hostname.endswith(".atlassian.net") or hostname == "atlassian.net": 

39 return ConfluenceDeploymentType.CLOUD 

40 return ConfluenceDeploymentType.DATACENTER 

41 except Exception: 

42 return ConfluenceDeploymentType.DATACENTER