Skip to content

Commit 2aeaf4a

Browse files
committed
MOD: Use standard library zstd in databento-python
1 parent 14640ba commit 2aeaf4a

14 files changed

Lines changed: 437 additions & 89 deletions

‎CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,11 @@
11
# Changelog
22

3+
## 0.88.0 - Upcoming
4+
5+
#### Enhancements
6+
- `zstandard` is now only a dependency on Python versions below 3.14; on Python 3.14 and
7+
later the standard library `compression.zstd` module is used instead
8+
39
## 0.87.0 - 2026-09-22
410

511
#### Enhancements

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ The minimum dependencies as found in the `pyproject.toml` are also listed below:
3838
- pip-system-certs >= 4.0 (Windows only)
3939
- pyarrow >= 13.0.0
4040
- requests >= 2.27.0
41-
- zstandard >= 0.21.0
41+
- zstandard >= 0.21.0 (Python < 3.14 only)
4242

4343
## Installation
4444
To install the latest stable version of the package from PyPI:

‎databento/common/dbnstore.py‎

Lines changed: 6 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@
3030
import pandas as pd
3131
import pyarrow as pa
3232
import pyarrow.parquet as pq
33-
import zstandard
3433
from databento_dbn import FIXED_PRICE_SCALE
3534
from databento_dbn import UNDEF_PRICE
3635
from databento_dbn import Compression
@@ -47,6 +46,7 @@
4746
from databento_dbn import Transcoder
4847
from databento_dbn import VersionUpgradePolicy
4948

49+
from databento.common import zstd
5050
from databento.common.constants import DEFINITION_TYPE_MAX_MAP
5151
from databento.common.constants import SCHEMA_STRUCT_MAP
5252
from databento.common.constants import SCHEMA_STRUCT_MAP_V1
@@ -86,12 +86,7 @@ def is_zstandard(reader: IO[bytes]) -> bool:
8686
8787
"""
8888
reader.seek(0) # ensure we read from the beginning
89-
try:
90-
zstandard.get_frame_parameters(reader.read(18))
91-
except zstandard.ZstdError:
92-
return False
93-
else:
94-
return True
89+
return zstd.is_frame_header(reader.read(18))
9590

9691

9792
def is_dbn(reader: IO[bytes]) -> bool:
@@ -356,7 +351,7 @@ def __init__(self, data_source: DataSource) -> None:
356351

357352
if is_zstandard(buffer):
358353
self._compression = Compression.ZSTD
359-
buffer = zstandard.ZstdDecompressor().stream_reader(data_source.reader)
354+
buffer = zstd.stream_reader(data_source.reader)
360355
elif is_dbn(buffer):
361356
self._compression = Compression.NONE
362357
buffer = data_source.reader
@@ -540,10 +535,7 @@ def reader(self) -> IO[bytes]:
540535
541536
"""
542537
if self.compression == Compression.ZSTD:
543-
return zstandard.ZstdDecompressor().stream_reader(
544-
self._data_source.reader,
545-
read_across_frames=True,
546-
)
538+
return zstd.stream_reader(self._data_source.reader)
547539
return self._data_source.reader
548540

549541
@property
@@ -1102,7 +1094,7 @@ def to_file(
11021094
compression = validate_maybe_enum(compression, Compression, "compression")
11031095
file_path = validate_file_write_path(path, "path", exist_ok=mode == "w")
11041096

1105-
writer: IO[bytes] | zstandard.ZstdCompressionWriter
1097+
writer: IO[bytes]
11061098
if compression is None or compression == self.compression:
11071099
# Handle trivial case
11081100
with open(file_path, mode=f"{mode}b") as writer:
@@ -1112,12 +1104,7 @@ def to_file(
11121104
return
11131105

11141106
if compression == Compression.ZSTD:
1115-
writer = zstandard.ZstdCompressor(
1116-
write_checksum=True,
1117-
).stream_writer(
1118-
open(file_path, mode=f"{mode}b"),
1119-
closefd=True,
1120-
)
1107+
writer = zstd.stream_writer(file_path, mode)
11211108
else:
11221109
writer = open(file_path, mode=f"{mode}b")
11231110

‎databento/common/parsing.py‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -12,9 +12,9 @@
1212
from typing import Any
1313

1414
import pandas as pd
15-
import zstandard
1615
from databento_dbn import SType
1716

17+
from databento.common import zstd
1818
from databento.common.constants import ALL_SYMBOLS
1919
from databento.common.enums import JobState
2020
from databento.common.validation import validate_enum
@@ -474,8 +474,7 @@ def convert_jsonl_to_df(data: bytes, compressed: bool) -> pd.DataFrame:
474474
475475
"""
476476
if compressed:
477-
decompressor = zstandard.ZstdDecompressor()
478-
reader: IO[bytes] = decompressor.stream_reader(data)
477+
reader: IO[bytes] = zstd.stream_reader(BytesIO(data))
479478
else:
480479
reader = BytesIO(data)
481480

‎databento/common/zstd.py‎

Lines changed: 255 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,255 @@
1+
from __future__ import annotations
2+
3+
import sys
4+
from io import BytesIO
5+
from os import PathLike
6+
from typing import IO
7+
from typing import Literal
8+
from typing import Protocol
9+
from typing import cast
10+
11+
12+
class IncrementalCompressor(Protocol):
13+
"""
14+
A compressor that accepts uncompressed data in chunks.
15+
"""
16+
17+
def compress(self, data: bytes) -> bytes: ...
18+
19+
def flush(self) -> bytes: ...
20+
21+
22+
if sys.version_info >= (3, 14):
23+
from compression import zstd as _zstd
24+
25+
ZstdError = _zstd.ZstdError
26+
27+
def is_frame_header(header: bytes) -> bool:
28+
"""
29+
Determine if `header` is the start of a zstandard frame.
30+
31+
Parameters
32+
----------
33+
header : bytes
34+
The first bytes of the data to check, at least 18 bytes to cover
35+
the largest frame header.
36+
37+
Returns
38+
-------
39+
bool
40+
41+
"""
42+
try:
43+
_zstd.get_frame_info(header)
44+
except _zstd.ZstdError:
45+
return False
46+
return True
47+
48+
def stream_reader(reader: IO[bytes]) -> IO[bytes]:
49+
"""
50+
Return a reader that decompresses zstandard data from `reader`.
51+
52+
The returned reader decompresses across all frames in `reader`.
53+
54+
Parameters
55+
----------
56+
reader : IO[bytes]
57+
The compressed data to read from.
58+
59+
Returns
60+
-------
61+
IO[bytes]
62+
63+
"""
64+
return cast(IO[bytes], _zstd.ZstdFile(reader, mode="rb"))
65+
66+
def stream_writer(
67+
path: PathLike[str] | str,
68+
mode: Literal["w", "x"],
69+
) -> IO[bytes]:
70+
"""
71+
Open the file at `path` for writing zstandard compressed data.
72+
73+
Closing the returned writer finishes the frame and closes the file.
74+
Frames are written with a content checksum.
75+
76+
Parameters
77+
----------
78+
path : PathLike[str] or str
79+
The file path to write to.
80+
mode : str
81+
The file write mode to use, either "x" or "w".
82+
83+
Returns
84+
-------
85+
IO[bytes]
86+
87+
"""
88+
return cast(
89+
IO[bytes],
90+
_zstd.ZstdFile(
91+
path,
92+
mode="wb" if mode == "w" else "xb",
93+
options={_zstd.CompressionParameter.checksum_flag: 1},
94+
),
95+
)
96+
97+
def compress(data: bytes) -> bytes:
98+
"""
99+
Compress `data` into a single zstandard frame.
100+
101+
Parameters
102+
----------
103+
data : bytes
104+
The data to compress.
105+
106+
Returns
107+
-------
108+
bytes
109+
110+
"""
111+
return _zstd.compress(data)
112+
113+
def decompress(data: bytes) -> bytes:
114+
"""
115+
Decompress all zstandard frames in `data`.
116+
117+
Parameters
118+
----------
119+
data : bytes
120+
The compressed data.
121+
122+
Returns
123+
-------
124+
bytes
125+
126+
"""
127+
return _zstd.decompress(data)
128+
129+
def incremental_compressor() -> IncrementalCompressor:
130+
"""
131+
Create a compressor that accepts uncompressed data in chunks.
132+
133+
Returns
134+
-------
135+
IncrementalCompressor
136+
137+
"""
138+
return _zstd.ZstdCompressor()
139+
140+
else:
141+
import zstandard
142+
143+
ZstdError = zstandard.ZstdError
144+
145+
def is_frame_header(header: bytes) -> bool:
146+
"""
147+
Determine if `header` is the start of a zstandard frame.
148+
149+
Parameters
150+
----------
151+
header : bytes
152+
The first bytes of the data to check, at least 18 bytes to cover
153+
the largest frame header.
154+
155+
Returns
156+
-------
157+
bool
158+
159+
"""
160+
try:
161+
zstandard.get_frame_parameters(header)
162+
except zstandard.ZstdError:
163+
return False
164+
return True
165+
166+
def stream_reader(reader: IO[bytes]) -> IO[bytes]:
167+
"""
168+
Return a reader that decompresses zstandard data from `reader`.
169+
170+
The returned reader decompresses across all frames in `reader`.
171+
172+
Parameters
173+
----------
174+
reader : IO[bytes]
175+
The compressed data to read from.
176+
177+
Returns
178+
-------
179+
IO[bytes]
180+
181+
"""
182+
return zstandard.ZstdDecompressor().stream_reader(
183+
reader,
184+
read_across_frames=True,
185+
)
186+
187+
def stream_writer(
188+
path: PathLike[str] | str,
189+
mode: Literal["w", "x"],
190+
) -> IO[bytes]:
191+
"""
192+
Open the file at `path` for writing zstandard compressed data.
193+
194+
Closing the returned writer finishes the frame and closes the file.
195+
Frames are written with a content checksum.
196+
197+
Parameters
198+
----------
199+
path : PathLike[str] or str
200+
The file path to write to.
201+
mode : str
202+
The file write mode to use, either "x" or "w".
203+
204+
Returns
205+
-------
206+
IO[bytes]
207+
208+
"""
209+
return zstandard.ZstdCompressor(write_checksum=True).stream_writer(
210+
open(path, mode=f"{mode}b"),
211+
closefd=True,
212+
)
213+
214+
def compress(data: bytes) -> bytes:
215+
"""
216+
Compress `data` into a single zstandard frame.
217+
218+
Parameters
219+
----------
220+
data : bytes
221+
The data to compress.
222+
223+
Returns
224+
-------
225+
bytes
226+
227+
"""
228+
return zstandard.compress(data)
229+
230+
def decompress(data: bytes) -> bytes:
231+
"""
232+
Decompress all zstandard frames in `data`.
233+
234+
Parameters
235+
----------
236+
data : bytes
237+
The compressed data.
238+
239+
Returns
240+
-------
241+
bytes
242+
243+
"""
244+
return stream_reader(BytesIO(data)).read()
245+
246+
def incremental_compressor() -> IncrementalCompressor:
247+
"""
248+
Create a compressor that accepts uncompressed data in chunks.
249+
250+
Returns
251+
-------
252+
IncrementalCompressor
253+
254+
"""
255+
return zstandard.ZstdCompressor().compressobj()

‎pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ dependencies = [
3131
"pip-system-certs>=4.0; platform_system == 'Windows'",
3232
"pyarrow>=13.0.0",
3333
"requests>=2.27.0",
34-
"zstandard>=0.21.0",
34+
"zstandard>=0.21.0; python_version < '3.14'",
3535
]
3636

3737
[project.urls]

0 commit comments

Comments
 (0)