|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from dataclasses import dataclass, field |
| 4 | +from typing import List |
| 5 | + |
| 6 | +_BRACES = {"(", ")", "[", "]", "{", "}"} |
| 7 | + |
| 8 | + |
| 9 | +def _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 | + |
| 24 | +def _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 | + |
| 47 | +def _validate_driver_version(version: str) -> None: |
| 48 | + _validate_no_invalid_chars(version, "Driver version") |
| 49 | + |
| 50 | + |
| 51 | +def _format_driver_entry(driver_name: str, driver_version: str) -> str: |
| 52 | + return f"{driver_name}_v{driver_version}" |
| 53 | + |
| 54 | + |
| 55 | +@dataclass |
| 56 | +class DriverInfo: |
| 57 | + """Driver information used to build the CLIENT SETINFO LIB-NAME value. |
| 58 | +
|
| 59 | + The formatted name follows the pattern:: |
| 60 | +
|
| 61 | + name(driver1_vVersion1;driver2_vVersion2) |
| 62 | +
|
| 63 | + Examples |
| 64 | + -------- |
| 65 | + >>> info = DriverInfo() |
| 66 | + >>> info.formatted_name |
| 67 | + 'redis-py' |
| 68 | +
|
| 69 | + >>> info = DriverInfo().add_upstream_driver("django-redis", "5.4.0") |
| 70 | + >>> info.formatted_name |
| 71 | + 'redis-py(django-redis_v5.4.0)' |
| 72 | + """ |
| 73 | + |
| 74 | + name: str = "redis-py" |
| 75 | + _upstream: List[str] = field(default_factory=list) |
| 76 | + |
| 77 | + @property |
| 78 | + def upstream_drivers(self) -> List[str]: |
| 79 | + """Return a copy of the upstream driver entries. |
| 80 | +
|
| 81 | + Each entry is in the form ``"driver-name_vversion"``. |
| 82 | + """ |
| 83 | + |
| 84 | + return list(self._upstream) |
| 85 | + |
| 86 | + def add_upstream_driver( |
| 87 | + self, driver_name: str, driver_version: str |
| 88 | + ) -> "DriverInfo": |
| 89 | + """Add an upstream driver to this instance and return self. |
| 90 | +
|
| 91 | + The most recently added driver appears first in :pyattr:`formatted_name`. |
| 92 | + """ |
| 93 | + |
| 94 | + if driver_name is None: |
| 95 | + raise ValueError("Driver name must not be None") |
| 96 | + if driver_version is None: |
| 97 | + raise ValueError("Driver version must not be None") |
| 98 | + |
| 99 | + _validate_driver_name(driver_name) |
| 100 | + _validate_driver_version(driver_version) |
| 101 | + |
| 102 | + entry = _format_driver_entry(driver_name, driver_version) |
| 103 | + # insert at the beginning so latest is first |
| 104 | + self._upstream.insert(0, entry) |
| 105 | + return self |
| 106 | + |
| 107 | + @property |
| 108 | + def formatted_name(self) -> str: |
| 109 | + """Return the base name with upstream drivers encoded, if any. |
| 110 | +
|
| 111 | + With no upstream drivers, this is just :pyattr:`name`. Otherwise:: |
| 112 | +
|
| 113 | + name(driver1_vX;driver2_vY) |
| 114 | + """ |
| 115 | + |
| 116 | + if not self._upstream: |
| 117 | + return self.name |
| 118 | + return f"{self.name}({';'.join(self._upstream)})" |
0 commit comments