Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/driver_info.py: 36%

1from __future__ import annotations 

2 

3from dataclasses import dataclass, field 

4from typing import List, Optional 

5 

6from redis.utils import SENTINEL 

7 

8_BRACES = {"(", ")", "[", "]", "{", "}"} 

9 

10 

11def _validate_no_invalid_chars(value: str, field_name: str) -> None: 

12 """Ensure value contains only printable ASCII without spaces or braces. 

13 

14 This mirrors the constraints enforced by other Redis clients for values that 

15 will appear in CLIENT LIST / CLIENT INFO output. 

16 """ 

17 

18 for ch in value: 

19 # printable ASCII without space: '!' (0x21) to '~' (0x7E) 

20 if ord(ch) < 0x21 or ord(ch) > 0x7E or ch in _BRACES: 

21 raise ValueError( 

22 f"{field_name} must not contain spaces, newlines, non-printable characters, or braces" 

23 ) 

24 

25 

26def _validate_driver_name(name: str) -> None: 

27 """Validate an upstream driver name. 

28 

29 The name should look like a typical Python distribution or package name, 

30 following a simplified form of PEP 503 normalisation rules: 

31 

32 * start with a lowercase ASCII letter 

33 * contain only lowercase letters, digits, hyphens and underscores 

34 

35 Examples of valid names: ``"django-redis"``, ``"celery"``, ``"rq"``. 

36 """ 

37 

38 import re 

39 

40 _validate_no_invalid_chars(name, "Driver name") 

41 if not re.match(r"^[a-z][a-z0-9_-]*$", name): 

42 raise ValueError( 

43 "Upstream driver name must use a Python package-style name: " 

44 "start with a lowercase letter and contain only lowercase letters, " 

45 "digits, hyphens, and underscores (e.g., 'django-redis')." 

46 ) 

47 

48 

49def _validate_driver_version(version: str) -> None: 

50 _validate_no_invalid_chars(version, "Driver version") 

51 

52 

53def _format_driver_entry(driver_name: str, driver_version: str) -> str: 

54 return f"{driver_name}_v{driver_version}" 

55 

56 

57@dataclass 

58class DriverInfo: 

59 """Driver information used to build the CLIENT SETINFO LIB-NAME and LIB-VER values. 

60 

61 This class consolidates all driver metadata (redis-py version and upstream drivers) 

62 into a single object that is propagated through connection pools and connections. 

63 

64 The formatted name follows the pattern:: 

65 

66 name(driver1_vVersion1;driver2_vVersion2) 

67 

68 Parameters 

69 ---------- 

70 name : str, optional 

71 The base library name. If omitted, defaults to "redis-py". If None, 

72 LIB-NAME will not be sent. 

73 lib_version : str, optional 

74 The redis-py library version. If omitted, the version will be determined 

75 automatically from the installed package. If None, LIB-VER will not be sent. 

76 

77 Examples 

78 -------- 

79 >>> info = DriverInfo() 

80 >>> info.formatted_name 

81 'redis-py' 

82 

83 >>> info = DriverInfo().add_upstream_driver("django-redis", "5.4.0") 

84 >>> info.formatted_name 

85 'redis-py(django-redis_v5.4.0)' 

86 

87 >>> info = DriverInfo(lib_version="5.0.0") 

88 >>> info.lib_version 

89 '5.0.0' 

90 """ 

91 

92 name: Optional[str] | object = SENTINEL 

93 lib_version: Optional[str] | object = SENTINEL 

94 _upstream: List[str] = field(default_factory=list) 

95 

96 def __post_init__(self): 

97 """Initialize default metadata if not explicitly provided.""" 

98 if self.name is SENTINEL: 

99 self.name = "redis-py" 

100 if self.lib_version is SENTINEL: 

101 from redis.utils import get_lib_version 

102 

103 self.lib_version = get_lib_version() 

104 

105 @property 

106 def upstream_drivers(self) -> List[str]: 

107 """Return a copy of the upstream driver entries. 

108 

109 Each entry is in the form ``"driver-name_vversion"``. 

110 """ 

111 

112 return list(self._upstream) 

113 

114 def add_upstream_driver( 

115 self, driver_name: str, driver_version: str 

116 ) -> "DriverInfo": 

117 """Add an upstream driver to this instance and return self. 

118 

119 The most recently added driver appears first in :pyattr:`formatted_name`. 

120 """ 

121 

122 if driver_name is None: 

123 raise ValueError("Driver name must not be None") 

124 if driver_version is None: 

125 raise ValueError("Driver version must not be None") 

126 

127 _validate_driver_name(driver_name) 

128 _validate_driver_version(driver_version) 

129 

130 entry = _format_driver_entry(driver_name, driver_version) 

131 # insert at the beginning so latest is first 

132 self._upstream.insert(0, entry) 

133 return self 

134 

135 @property 

136 def formatted_name(self) -> Optional[str]: 

137 """Return the base name with upstream drivers encoded, if any. 

138 

139 With no upstream drivers, this is just :pyattr:`name`. Otherwise:: 

140 

141 name(driver1_vX;driver2_vY) 

142 """ 

143 

144 name = self.name 

145 if not isinstance(name, str) or not name: 

146 return None 

147 if not self._upstream: 

148 return name 

149 return f"{name}({';'.join(self._upstream)})" 

150 

151 

152def resolve_driver_info( 

153 driver_info: Optional[DriverInfo] | object = SENTINEL, 

154 lib_name: Optional[str] | object = SENTINEL, 

155 lib_version: Optional[str] | object = SENTINEL, 

156) -> Optional[DriverInfo]: 

157 """Resolve driver_info from parameters. 

158 

159 If driver_info is provided, use it. Otherwise, create DriverInfo from 

160 lib_name and lib_version (using defaults only for sentinel values). 

161 

162 Parameters 

163 ---------- 

164 driver_info : DriverInfo, optional 

165 The DriverInfo instance to use 

166 lib_name : str, optional 

167 The library name (default: "redis-py") 

168 lib_version : str, optional 

169 The library version (default: auto-detected) 

170 

171 Returns 

172 ------- 

173 DriverInfo, optional 

174 The resolved DriverInfo instance 

175 """ 

176 if driver_info is SENTINEL: 

177 if lib_name is None and lib_version is None: 

178 return None 

179 return DriverInfo(name=lib_name, lib_version=lib_version) 

180 

181 if driver_info is None or isinstance(driver_info, DriverInfo): 

182 return driver_info 

183 

184 raise TypeError("driver_info must be a DriverInfo instance or None")