Coverage for src / qdrant_loader / cli / commands / webhook_cmd.py: 0%

1from __future__ import annotations 

2 

3from pathlib import Path 

4 

5import uvicorn 

6from click.exceptions import ClickException 

7 

8from qdrant_loader.cli.config_loader import ( 

9 load_config_with_workspace, 

10 setup_workspace, 

11) 

12from qdrant_loader.config.workspace import validate_workspace_flags 

13from qdrant_loader.utils.logging import LoggingConfig 

14from qdrant_loader.utils.sensitive import sanitize_exception_message 

15from qdrant_loader.webhooks.server import app 

16 

17 

18def _setup_logging(log_level: str, workspace_config) -> None: 

19 log_file = ( 

20 str(workspace_config.logs_path / "webhook.log") 

21 if workspace_config 

22 else "qdrant-loader-webhook.log" 

23 ) 

24 if getattr(LoggingConfig, "reconfigure", None): 

25 if getattr(LoggingConfig, "_initialized", False): 

26 LoggingConfig.reconfigure(file=log_file, level=log_level) 

27 else: 

28 LoggingConfig.setup(level=log_level, format="console", file=log_file) 

29 else: 

30 LoggingConfig.setup(level=log_level, format="console", file=log_file) 

31 

32 

33async def run_webhook_command( 

34 workspace: Path | None, 

35 config: Path | None, 

36 env: Path | None, 

37 host: str, 

38 port: int, 

39 log_level: str, 

40) -> None: 

41 """Run the webhook server for connector events.""" 

42 try: 

43 validate_workspace_flags(workspace, config, env) 

44 workspace_config = setup_workspace(workspace) if workspace else None 

45 except ValueError as exc: 

46 raise ClickException(str(exc)) from exc 

47 

48 _setup_logging(log_level, workspace_config) 

49 

50 try: 

51 load_config_with_workspace(workspace_config, config, env) 

52 except Exception as exc: 

53 safe_error = sanitize_exception_message(exc) or type(exc).__name__ 

54 raise ClickException(f"Failed to load configuration: {safe_error}") from exc 

55 

56 logger = LoggingConfig.get_logger(__name__) 

57 logger.info( 

58 "Starting webhook server", 

59 host=host, 

60 port=port, 

61 workspace=str(workspace_config.workspace_path) if workspace_config else None, 

62 ) 

63 

64 server_config = uvicorn.Config( 

65 app, 

66 host=host, 

67 port=port, 

68 log_level=log_level.lower(), 

69 ) 

70 try: 

71 server = uvicorn.Server(server_config) 

72 await server.serve() 

73 except Exception as exc: 

74 logger.exception("Webhook server error") 

75 safe_error = sanitize_exception_message(exc) or type(exc).__name__ 

76 raise ClickException(f"Webhook server failed: {safe_error}") from exc