add bzip2 codec
MODIFY
README.md
README.md
@@ -32,7 +32,7 @@ encoding/decoding).
* Schemaless Reader
* JSON Writer
* JSON Reader
-* Codecs (Snappy, Deflate, Zstandard)
+* Codecs (Snappy, Deflate, Zstandard, Bzip2)
* Schema resolution
* Aliases
* Logical Types

MODIFY
docs/index.rst
docs/index.rst
@@ -27,7 +27,7 @@ Supported Features
* Schemaless Reader
* JSON Writer
* JSON Reader
-* Codecs (Snappy, Deflate, Zstandard)
+* Codecs (Snappy, Deflate, Zstandard, Bzip2)
* Schema resolution
* Aliases
* Logical Types

MODIFY
fastavro/_read.pyx
fastavro/_read.pyx
# http://svn.apache.org/viewvc/avro/trunk/lang/py/src/avro/ which is under
# Apache 2.0 license (http://www.apache.org/licenses/LICENSE-2.0)
-from zlib import decompress
+import bz2
+import zlib
import datetime
from decimal import localcontext, Decimal
from fastavro.six import MemoryIO
@@ -602,12 +603,19 @@ cpdef deflate_read_block(fo):
data = read_bytes(fo)
# -15 is the log of the window size; negative indicates "raw" (no
# zlib headers) decompression. See zlib.h.
- return MemoryIO(decompress(data, -15))
+ return MemoryIO(zlib.decompress(data, -15))
+
+
+cpdef bzip2_read_block(fo):
+ """Read block in "bzip2" codec."""
+ data = read_bytes(fo)
+ return MemoryIO(bz2.decompress(data))
BLOCK_READERS = {
'null': null_read_block,
- 'deflate': deflate_read_block
+ 'deflate': deflate_read_block,
+ 'bzip2': bzip2_read_block,
}

MODIFY
fastavro/_read_py.py
fastavro/_read_py.py
from fastavro.six import MemoryIO
from struct import error as StructError
-from zlib import decompress
+import bz2
+import zlib
import datetime
from decimal import localcontext, Decimal
from uuid import UUID
@@ -462,12 +463,19 @@ def deflate_read_block(decoder):
data = decoder.read_bytes()
# -15 is the log of the window size; negative indicates "raw" (no
# zlib headers) decompression. See zlib.h.
- return MemoryIO(decompress(data, -15))
+ return MemoryIO(zlib.decompress(data, -15))
+
+
+def bzip2_read_block(decoder):
+ """Read block in "bzip2" codec."""
+ data = decoder.read_bytes()
+ return MemoryIO(bz2.decompress(data))
BLOCK_READERS = {
'null': null_read_block,
- 'deflate': deflate_read_block
+ 'deflate': deflate_read_block,
+ 'bzip2': bzip2_read_block,
}

MODIFY
fastavro/_write.pyx
fastavro/_write.pyx
import json
from binascii import crc32
from os import urandom
-from zlib import compress
+import bz2
+import zlib
from fastavro import const
from ._logical_writers import LOGICAL_WRITERS
@@ -358,16 +359,26 @@ cpdef deflate_write_block(object fo, bytes block_bytes):
cdef bytearray tmp = bytearray()
# The first two characters and last character are zlib
# wrappers around deflate data.
- data = compress(block_bytes)[2:-1]
+ data = zlib.compress(block_bytes)[2:-1]
write_long(tmp, len(data))
fo.write(tmp)
fo.write(data)
+cpdef bzip2_write_block(object fo, bytes block_bytes):
+ """Write block in "bzip2" codec."""
+ cdef bytearray tmp = bytearray()
+ data = bz2.compress(block_bytes)
+ write_long(tmp, len(data))
+ fo.write(tmp)
+ fo.write(data)
+
+
BLOCK_WRITERS = {
'null': null_write_block,
- 'deflate': deflate_write_block
+ 'deflate': deflate_write_block,
+ 'bzip2': bzip2_write_block,
}

MODIFY
fastavro/_write_py.py
fastavro/_write_py.py
import json
from os import urandom, SEEK_SET
-from zlib import compress
+import bz2
+import zlib
from .io.binary_encoder import BinaryEncoder
from .io.json_encoder import AvroJSONEncoder
@@ -253,15 +254,22 @@ def deflate_write_block(encoder, block_bytes):
"""Write block in "deflate" codec."""
# The first two characters and last character are zlib
# wrappers around deflate data.
- data = compress(block_bytes)[2:-1]
+ data = zlib.compress(block_bytes)[2:-1]
+ encoder.write_long(len(data))
+ encoder._fo.write(data)
+
+def bzip2_write_block(encoder, block_bytes):
+ """Write block in "bzip2" codec."""
+ data = bz2.compress(block_bytes)
encoder.write_long(len(data))
encoder._fo.write(data)
BLOCK_WRITERS = {
'null': null_write_block,
- 'deflate': deflate_write_block
+ 'deflate': deflate_write_block,
+ 'bzip2': bzip2_write_block,
}

MODIFY
tests/test_compression.py
tests/test_compression.py
@@ -6,9 +6,38 @@ from fastavro.six import MemoryIO
import fastavro
+@pytest.mark.parametrize("codec", ["null", "deflate", "bzip2"])
+def test_builtin_codecs(codec):
+ schema = {
+ "doc": "A weather reading.",
+ "name": "Weather",
+ "namespace": "test",
+ "type": "record",
+ "fields": [
+ {"name": "station", "type": "string"},
+ {"name": "time", "type": "long"},
+ {"name": "temp", "type": "int"},
+ ],
+ }
+
+ records = [
+ {"station": "011990-99999", "temp": 0, "time": 1433269388},
+ {"station": "011990-99999", "temp": 22, "time": 1433270389},
+ {"station": "011990-99999", "temp": -11, "time": 1433273379},
+ {"station": "012650-99999", "temp": 111, "time": 1433275478},
+ ]
+
+ file = MemoryIO()
+ fastavro.writer(file, schema, records, codec=codec)
+
+ file.seek(0)
+ out_records = list(fastavro.reader(file))
+ assert records == out_records
+
+
@pytest.mark.parametrize("codec", ["snappy", "zstandard"])
@pytest.mark.skipif(os.name == "nt", reason="A pain to set up on windows")
-def test_snappy(codec):
+def test_optional_codecs(codec):
schema = {
"doc": "A weather reading.",
"name": "Weather",
@@ -38,7 +67,7 @@ def test_snappy(codec):
@pytest.mark.parametrize("codec", ["snappy", "zstandard"])
@pytest.mark.skipif(os.name != "nt", reason="codec is present")
-def test_snappy_not_installed(codec):
+def test_optional_codecs_not_installed(codec):
schema = {
"doc": "A weather reading.",
"name": "Weather",

add bzip2 codec