Coverage for seedboxsync/core/flask.py: 100%

1# 

2# Copyright (C) 2015-2026 Guillaume Kulakowski <guillaume@kulakowski.fr> 

3# 

4# For the full copyright and license information, please view the LICENSE 

5# file that was distributed with this source code. 

6# 

7"""Flask core initer mobule.""" 

8 

9from functools import cached_property 

10from importlib import import_module 

11from typing import Any, cast 

12from flask import Flask, current_app 

13from seedboxsync.core import Config 

14from seedboxsync.core.exception import PingServiceError, SyncProtocoleError 

15from seedboxsync.core.ping import AbstractPingClient 

16from seedboxsync.core.sync import AbstractSyncClient 

17from seedboxsync.core.taskmanager import Manager, task_manager 

18 

19 

20class SeedboxSyncFlask(Flask): 

21 """Flask application with SeedboxSync-specific configuration helpers.""" 

22 

23 @property 

24 def seedboxsync_config(self) -> dict[str, Any]: 

25 """ 

26 Return the SeedboxSync configuration namespace. 

27 

28 Returns: 

29 The SeedboxSync configuration with the namespace prefix removed 

30 and keys converted to lowercase. 

31 """ 

32 return self.config.get_namespace(Config.CONFIG_NAMESPACE) 

33 

34 @cached_property 

35 def task_manager(self) -> Manager: 

36 """ 

37 Return the task instance Manager. 

38 

39 Returns: 

40 The task manager instance. 

41 """ 

42 task_manager.init_app(self) 

43 return task_manager 

44 

45 @cached_property 

46 def sync(self) -> AbstractSyncClient: 

47 """ 

48 Return the configured sync client instance. 

49 

50 Returns: 

51 AbstractSyncClient: Client instance. 

52 """ 

53 protocol = self.seedboxsync_config.get("seedbox_protocol", "") 

54 client_class = protocol.title() + "Client" 

55 

56 self.logger.debug(f"Using SeedboxSync with sync ({protocol} / {client_class})") 

57 

58 # Load the client module dynamically 

59 try: 

60 client_module = import_module("seedboxsync.core.sync.client." + protocol) 

61 except ImportError as exc: 

62 raise SyncProtocoleError(f"Unsupported protocol: {protocol}: {exc!s}") from exc 

63 

64 # Get the client class from the module 

65 try: 

66 transfer_client = getattr(client_module, client_class) 

67 except AttributeError as exc: 

68 raise SyncProtocoleError(f'Unsupported protocol module! No class "{client_class}" in module "seedboxsync.core.sync.{protocol}_client"') from exc 

69 

70 # Instantiate the client 

71 try: 

72 sync_client = transfer_client() 

73 except Exception as exc: 

74 raise ConnectionError(f"{exc!s}\nFailed to establish a connection.") from exc 

75 

76 return sync_client # type: ignore[no-any-return] 

77 

78 @cached_property 

79 def ping(self) -> AbstractPingClient: 

80 """ 

81 Return the configured ping client instance. 

82 

83 Returns: 

84 AbstractPingClient: Configured ping client instance. 

85 """ 

86 ping_service = "healthchecks" 

87 client_class = ping_service.title() 

88 

89 self.logger.debug(f"Using SeedboxSync with ping ({ping_service} / {client_class})") 

90 

91 # Load the client module dynamically 

92 try: 

93 client_module = import_module("seedboxsync.core.ping.client." + ping_service) 

94 except ImportError as exc: 

95 raise PingServiceError(f"Unsupported ping service: {ping_service}: {exc!s}") from exc 

96 

97 # Get the client class from the module 

98 try: 

99 ping_client = getattr(client_module, client_class) 

100 except AttributeError as exc: 

101 raise PingServiceError( 

102 f'Unsupported ping service module! No class "{client_class}" in module "seedboxsync.core.ping.{ping_service}client"' 

103 ) from exc 

104 

105 return ping_client() # type: ignore[no-any-return] 

106 

107 

108seedboxsync_current_app = cast(SeedboxSyncFlask, current_app)