Coverage for src/mafw/db/db_connection.py: 100%

1# Copyright 2026 European Union 

2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu) 

3# SPDX-License-Identifier: EUPL-1.2 

4""" 

5Database connection configuration helpers. 

6 

7:Author: Bulgheroni Antonio 

8:Description: Normalize database steering configuration into peewee connection arguments. 

9""" 

10 

11from __future__ import annotations 

12 

13import os 

14import warnings 

15from collections.abc import Mapping 

16from typing import Any 

17 

18from mafw.mafw_errors import SteeringFileDeprecation 

19from mafw.tools.regexp import extract_protocol 

20 

21VALID_AUTH_METHODS: set[str] = {'env', 'inline', 'file'} 

22"""Supported authentication methods for ``DBConfiguration.authentication``.""" 

23 

24LEGACY_KEYS: set[str] = {'URL', 'pragmas'} 

25"""Keys reserved by the legacy DBConfiguration schema.""" 

26 

27 

28def build_connection_parameters(config: Mapping[str, Any]) -> tuple[str, dict[str, Any]]: 

29 """Normalize DBConfiguration mappings into peewee connection arguments. 

30 

31 The input can either be the full steering configuration dictionary or the 

32 nested ``DBConfiguration`` table. The returned tuple provides the URL and 

33 the keyword arguments for ``peewee.connect``. 

34 

35 :param config: Steering configuration or DBConfiguration table. 

36 :type config: Mapping[str, Any] 

37 :return: Tuple with database URL and connection keyword arguments. 

38 :rtype: tuple[str, dict[str, Any]] 

39 :raises ValueError: When required fields are missing or invalid. 

40 """ 

41 

42 conf = dict(config) 

43 if 'DBConfiguration' in conf: 

44 conf = dict(conf['DBConfiguration']) 

45 

46 url = conf.get('URL') 

47 if not url: 

48 raise ValueError('DBConfiguration must define a URL field.') 

49 

50 protocol = extract_protocol(str(url)) or '' 

51 is_new_style = 'authentication' in conf or 'parameters' in conf 

52 

53 if not is_new_style: 

54 warnings.warn( 

55 'Legacy DBConfiguration schema detected. Please update to the new authentication/parameters layout.', 

56 SteeringFileDeprecation, 

57 stacklevel=2, 

58 ) 

59 return str(url), _legacy_connection_parameters(protocol, conf) 

60 

61 connection_parameters: dict[str, Any] = {} 

62 _apply_backend_parameters(protocol, conf, connection_parameters) 

63 _apply_authentication(protocol, conf.get('authentication'), connection_parameters) 

64 

65 return str(url), connection_parameters 

66 

67 

68def _legacy_connection_parameters(protocol: str, conf: Mapping[str, Any]) -> dict[str, Any]: 

69 connection_parameters: dict[str, Any] = {} 

70 if protocol == 'sqlite': 

71 connection_parameters['pragmas'] = conf.get('pragmas', {}) 

72 for key, value in conf.items(): 

73 if key in LEGACY_KEYS: 

74 continue 

75 connection_parameters[key] = value 

76 return connection_parameters 

77 

78 

79def _apply_backend_parameters(protocol: str, conf: Mapping[str, Any], target: dict[str, Any]) -> None: 

80 parameters = conf.get('parameters') 

81 if not isinstance(parameters, Mapping): 

82 return 

83 

84 backend_params = parameters.get(protocol) 

85 if not isinstance(backend_params, Mapping): 

86 return 

87 

88 backend_dict = dict(backend_params) 

89 if protocol == 'sqlite': 

90 pragmas = backend_dict.pop('pragmas', None) 

91 if isinstance(pragmas, Mapping): 

92 target['pragmas'] = dict(pragmas) 

93 for key, value in backend_dict.items(): 

94 target[key] = value 

95 

96 

97def _apply_authentication(protocol: str, auth: Any, target: dict[str, Any]) -> None: 

98 if protocol == 'sqlite': 

99 return 

100 

101 if auth is None or not isinstance(auth, Mapping): 

102 raise ValueError('DBConfiguration.authentication must be defined for server-based databases.') 

103 

104 method = auth.get('method') 

105 if method not in VALID_AUTH_METHODS: 

106 raise ValueError(f'Unsupported authentication method: {method!r}.') 

107 

108 if method == 'env': 

109 _apply_env_authentication(auth, target) 

110 return 

111 

112 if method == 'inline': 

113 warnings.warn( 

114 'Inline database credentials are insecure. Prefer environment-based authentication.', 

115 UserWarning, 

116 stacklevel=2, 

117 ) 

118 _apply_inline_authentication(auth, target) 

119 return 

120 

121 if method == 'file': 

122 _apply_file_authentication(protocol, auth, target) 

123 return 

124 

125 

126def _apply_env_authentication(auth: Mapping[str, Any], target: dict[str, Any]) -> None: 

127 if 'username' in auth and auth['username'] is not None: 

128 env_name = str(auth['username']) 

129 value = os.getenv(env_name) 

130 if value is None: 

131 raise ValueError(f'Environment variable {env_name!r} for DB username is not set.') 

132 target['user'] = value 

133 if 'password' in auth and auth['password'] is not None: 

134 env_name = str(auth['password']) 

135 value = os.getenv(env_name) 

136 if value is None: 

137 raise ValueError(f'Environment variable {env_name!r} for DB password is not set.') 

138 target['password'] = value 

139 

140 

141def _apply_inline_authentication(auth: Mapping[str, Any], target: dict[str, Any]) -> None: 

142 if 'username' in auth and auth['username'] is not None: 

143 target['user'] = auth['username'] 

144 if 'password' in auth and auth['password'] is not None: 

145 target['password'] = auth['password'] 

146 

147 

148def _apply_file_authentication(protocol: str, auth: Mapping[str, Any], target: dict[str, Any]) -> None: 

149 if 'username' in auth and auth['username'] is not None: 

150 target['user'] = auth['username'] 

151 

152 passfile = auth.get('passfile') 

153 if passfile is None: 

154 return 

155 

156 if protocol == 'mysql': 

157 target['read_default_file'] = passfile 

158 elif protocol == 'postgresql': 

159 target['passfile'] = passfile 

160 else: 

161 raise ValueError(f'Authentication method "file" is not supported for {protocol!r}.')