(dummy: Any, value: Union[str, Iterable[str]])
| 56 | |
| 57 | |
| 58 | def validate_compressors(dummy: Any, value: Union[str, Iterable[str]]) -> list[str]: |
| 59 | try: |
| 60 | # `value` is string. |
| 61 | compressors = value.split(",") # type: ignore[union-attr] |
| 62 | except AttributeError: |
| 63 | # `value` is an iterable. |
| 64 | compressors = list(value) |
| 65 | |
| 66 | for compressor in compressors[:]: |
| 67 | if compressor not in _SUPPORTED_COMPRESSORS: |
| 68 | compressors.remove(compressor) |
| 69 | warnings.warn(f"Unsupported compressor: {compressor}", stacklevel=2) |
| 70 | elif compressor == "snappy" and not _have_snappy(): |
| 71 | compressors.remove(compressor) |
| 72 | warnings.warn( |
| 73 | "Wire protocol compression with snappy is not available. " |
| 74 | "You must install the python-snappy module for snappy support.", |
| 75 | stacklevel=2, |
| 76 | ) |
| 77 | elif compressor == "zlib" and not _have_zlib(): |
| 78 | compressors.remove(compressor) |
| 79 | warnings.warn( |
| 80 | "Wire protocol compression with zlib is not available. " |
| 81 | "The zlib module is not available.", |
| 82 | stacklevel=2, |
| 83 | ) |
| 84 | elif compressor == "zstd" and not _have_zstd(): |
| 85 | compressors.remove(compressor) |
| 86 | if sys.version_info >= (3, 14): |
| 87 | warnings.warn( |
| 88 | "Wire protocol compression with zstandard is not available. " |
| 89 | "The compression.zstd module is not available.", |
| 90 | stacklevel=2, |
| 91 | ) |
| 92 | else: |
| 93 | warnings.warn( |
| 94 | "Wire protocol compression with zstandard is not available. " |
| 95 | "You must install the backports.zstd module for zstandard support.", |
| 96 | stacklevel=2, |
| 97 | ) |
| 98 | return compressors |
| 99 | |
| 100 | |
| 101 | def validate_zlib_compression_level(option: str, value: Any) -> int: |
nothing calls this directly
no test coverage detected