Use compression.zstd (PEP-784)

Author: Rogdham

kombu/compression.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import sys
 import zlib
 
 from kombu.utils.encoding import ensure_bytes
@@ -104,18 +105,13 @@ def decompress(body, content_type):
              'application/x-lzma', aliases=['lzma', 'xz'])
 
 try:
-    import zstandard as zstd
+    if sys.version_info >= (3, 14):
+        from compression import zstd
+    else:
+        from backports import zstd
 except ImportError:  # pragma: no cover
-    pass
+    pass  # no zstd support
 else:
-    def zstd_compress(body):
-        c = zstd.ZstdCompressor()
-        return c.compress(body)
-
-    def zstd_decompress(body):
-        d = zstd.ZstdDecompressor()
-        return d.decompress(body)
-
-    register(zstd_compress,
-             zstd_decompress,
+    register(zstd.compress,
+             zstd.decompress,
              'application/zstd', aliases=['zstd', 'zstandard'])

requirements/extras/zstd.txt

@@ -1 +1 @@
-zstandard==0.25.0
+backports.zstd==1.0.0; python_version<'3.14'

t/unit/test_compression.py

@@ -27,7 +27,7 @@ def test_encoders__lzma(self):
         assert 'application/x-lzma' in compression.encoders()
 
     def test_encoders__zstd(self):
-        pytest.importorskip('zstandard')
+        pytest.importorskip('compression.zstd' if sys.version_info >= (3, 14) else 'backports.zstd')
 
         assert 'application/zstd' in compression.encoders()
 
@@ -64,7 +64,7 @@ def test_compress__decompress__lzma(self):
         assert d == text
 
     def test_compress__decompress__zstd(self):
-        pytest.importorskip('zstandard')
+        pytest.importorskip('compression.zstd' if sys.version_info >= (3, 14) else 'backports.zstd')
 
         text = b'The Brown Quick Fox Over The Lazy Dog Jumps'
         c, ctype = compression.compress(text, 'zstd')