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

1from __future__ import annotations 

2 

3from dataclasses import dataclass, field 

4from typing import List, Optional 

5 

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

7 

8 

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

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

11 

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

13 will appear in CLIENT LIST / CLIENT INFO output. 

14 """ 

15 

16 for ch in value: 

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

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

19 raise ValueError( 

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

21 ) 

22 

23 

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

25 """Validate an upstream driver name. 

26 

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

28 following a simplified form of PEP 503 normalisation rules: 

29 

30 * start with a lowercase ASCII letter 

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

32 

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

34 """ 

35 

36 import re 

37 

38 _validate_no_invalid_chars(name, "Driver name") 

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

40 raise ValueError( 

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

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

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

44 ) 

45 

46 

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

48 _validate_no_invalid_chars(version, "Driver version") 

49 

50 

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

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

53 

54 

55@dataclass 

56class DriverInfo: 

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

58 

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

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

61 

62 The formatted name follows the pattern:: 

63 

64 name(driver1_vVersion1;driver2_vVersion2) 

65 

66 Parameters 

67 ---------- 

68 name : str, optional 

69 The base library name (default: "redis-py") 

70 lib_version : str, optional 

71 The redis-py library version. If None, the version will be determined 

72 automatically from the installed package. 

73 

74 Examples 

75 -------- 

76 >>> info = DriverInfo() 

77 >>> info.formatted_name 

78 'redis-py' 

79 

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

81 >>> info.formatted_name 

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

83 

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

85 >>> info.lib_version 

86 '5.0.0' 

87 """ 

88 

89 name: str = "redis-py" 

90 lib_version: Optional[str] = None 

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

92 

93 def __post_init__(self): 

94 """Initialize lib_version if not provided.""" 

95 if self.lib_version is None: 

96 from redis.utils import get_lib_version 

97 

98 self.lib_version = get_lib_version() 

99 

100 @property 

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

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

103 

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

105 """ 

106 

107 return list(self._upstream) 

108 

109 def add_upstream_driver( 

110 self, driver_name: str, driver_version: str 

111 ) -> "DriverInfo": 

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

113 

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

115 """ 

116 

117 if driver_name is None: 

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

119 if driver_version is None: 

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

121 

122 _validate_driver_name(driver_name) 

123 _validate_driver_version(driver_version) 

124 

125 entry = _format_driver_entry(driver_name, driver_version) 

126 # insert at the beginning so latest is first 

127 self._upstream.insert(0, entry) 

128 return self 

129 

130 @property 

131 def formatted_name(self) -> str: 

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

133 

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

135 

136 name(driver1_vX;driver2_vY) 

137 """ 

138 

139 if not self._upstream: 

140 return self.name 

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

142 

143 

144def resolve_driver_info( 

145 driver_info: Optional[DriverInfo], 

146 lib_name: Optional[str], 

147 lib_version: Optional[str], 

148) -> DriverInfo: 

149 """Resolve driver_info from parameters. 

150 

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

152 lib_name and lib_version (using defaults if not provided). 

153 

154 Parameters 

155 ---------- 

156 driver_info : DriverInfo, optional 

157 The DriverInfo instance to use 

158 lib_name : str, optional 

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

160 lib_version : str, optional 

161 The library version (default: auto-detected) 

162 

163 Returns 

164 ------- 

165 DriverInfo 

166 The resolved DriverInfo instance 

167 """ 

168 if driver_info is not None: 

169 return driver_info 

170 

171 # Fallback: create DriverInfo from lib_name and lib_version 

172 from redis.utils import get_lib_version 

173 

174 name = lib_name if lib_name is not None else "redis-py" 

175 version = lib_version if lib_version is not None else get_lib_version() 

176 return DriverInfo(name=name, lib_version=version)