add support for appending to files
MODIFY
fastavro/_six.pyx
fastavro/_six.pyx
@@ -45,6 +45,18 @@ if sys.version_info >= (3, 0):
def py3_fstint(datum):
return datum[0]
+ def py3_appendable(file_like):
+ if file_like.tell() != 0:
+ if file_like.readable():
+ return True
+ else:
+ raise ValueError(
+ "When appending to an avro file you must use the " +
+ "'a+' mode, not just 'a'"
+ )
+ else:
+ return False
+
else: # Python 2x
from cStringIO import StringIO as MemoryIO # flake8: noqa
xrange = xrange
@@ -84,6 +96,37 @@ else: # Python 2x
def py2_fstint(datum):
return unpack('!b', datum[0])[0]
+ def _readable(file_like):
+ try:
+ file_like.read()
+ except Exception:
+ return False
+ return True
+
+ def py2_appendable(file_like):
+ # On Python 2 things are a mess. We basically just rely on looking at
+ # the mode. If that doesn't exist (like in the case of an io.BytesIO)
+ # then we check the position and readablility.
+ try:
+ mode = file_like.mode
+ except AttributeError:
+ # This is probably some io stream so we rely on its tell() working
+ if file_like.tell() != 0 and _readable(file_like):
+ return True
+ else:
+ return False
+
+ if "a" in file_like.mode:
+ if "+" in file_like.mode:
+ return True
+ else:
+ raise ValueError(
+ "When appending to an avro file you must use the " +
+ "'a+' mode, not just 'a'"
+ )
+ else:
+ return False
+
# We do it this way and not just redifine function since Cython do not like it
if sys.version_info >= (3, 0):
btou = py3_btou
@@ -97,6 +140,7 @@ if sys.version_info >= (3, 0):
mk_bits = py3_mk_bits
str2ints = py3_bytes2ints
fstint = py3_fstint
+ appendable = py3_appendable
else:
btou = py2_btou
utob = py2_utob
@@ -109,3 +153,4 @@ else:
mk_bits = py2_mk_bits
str2ints = py2_str2ints
fstint = py2_fstint
+ appendable = py2_appendable

MODIFY
fastavro/_write.pyx
fastavro/_write.pyx
@@ -17,8 +17,8 @@ from zlib import compress
from fastavro import const
from ._validation import validate
-from ._six import utob, long, iteritems, mk_bits
-from ._read import HEADER_SCHEMA, SYNC_SIZE, MAGIC
+from ._six import utob, long, iteritems, mk_bits, appendable
+from ._read import HEADER_SCHEMA, SYNC_SIZE, MAGIC, reader
from ._schema import extract_record_type, extract_logical_type, parse_schema
from ._schema_common import SCHEMA_DEFS
from ._timezone import epoch
@@ -629,16 +629,39 @@ cdef class Writer(object):
metadata=None,
validator=None):
cdef bytearray tmp = bytearray()
+
self.fo = fo
self.schema = parse_schema(schema)
self.validate_fn = validate if validator is True else validator
- self.sync_marker = bytes(urandom(SYNC_SIZE))
self.io = MemoryIO()
self.block_count = 0
+ self.sync_interval = sync_interval
+
+ if appendable(self.fo):
+ # Seed to the beginning to read the header
+ self.fo.seek(0)
+ avro_reader = reader(self.fo)
+ header = avro_reader._header
+
+ file_writer_schema = parse_schema(avro_reader.writer_schema)
+ if self.schema != file_writer_schema:
+ msg = "Provided schema {} does not match file writer_schema {}"
+ raise ValueError(msg.format(self.schema, file_writer_schema))
+
+ codec = avro_reader.metadata.get("avro.codec", "null")
+
+ self.sync_marker = header["sync"]
+
+ # Seek to the end of the file
+ self.fo.seek(0, 2)
+
+ self.block_writer = BLOCK_WRITERS[codec]
+ else:
+ self.sync_marker = urandom(SYNC_SIZE)
+
self.metadata = metadata or {}
self.metadata['avro.codec'] = codec
self.metadata['avro.schema'] = json.dumps(schema)
- self.sync_interval = sync_interval
try:
self.block_writer = BLOCK_WRITERS[codec]

MODIFY
fastavro/_write_py.py
fastavro/_write_py.py
@@ -22,8 +22,8 @@ from .const import (
MCS_PER_HOUR, MCS_PER_MINUTE, MCS_PER_SECOND, MLS_PER_HOUR, MLS_PER_MINUTE,
MLS_PER_SECOND, DAYS_SHIFT
)
-from .six import utob, MemoryIO, long, iteritems, mk_bits
-from .read import HEADER_SCHEMA, SYNC_SIZE, MAGIC
+from .six import utob, MemoryIO, long, iteritems, mk_bits, appendable
+from .read import HEADER_SCHEMA, SYNC_SIZE, MAGIC, reader
from .schema import extract_record_type, extract_logical_type, parse_schema
from ._schema_common import SCHEMA_DEFS
from ._timezone import epoch
@@ -472,13 +472,35 @@ class Writer(object):
self.fo = fo
self.schema = parse_schema(schema)
self.validate_fn = validate if validator is True else validator
- self.sync_marker = urandom(SYNC_SIZE)
self.io = MemoryIO()
self.block_count = 0
+ self.sync_interval = sync_interval
+
+ if appendable(self.fo):
+ # Seed to the beginning to read the header
+ self.fo.seek(0)
+ avro_reader = reader(self.fo)
+ header = avro_reader._header
+
+ file_writer_schema = parse_schema(avro_reader.writer_schema)
+ if self.schema != file_writer_schema:
+ msg = "Provided schema {} does not match file writer_schema {}"
+ raise ValueError(msg.format(self.schema, file_writer_schema))
+
+ codec = avro_reader.metadata.get("avro.codec", "null")
+
+ self.sync_marker = header["sync"]
+
+ # Seek to the end of the file
+ self.fo.seek(0, 2)
+
+ self.block_writer = BLOCK_WRITERS[codec]
+ else:
+ self.sync_marker = urandom(SYNC_SIZE)
+
self.metadata = metadata or {}
self.metadata['avro.codec'] = codec
self.metadata['avro.schema'] = json.dumps(schema)
- self.sync_interval = sync_interval
try:
self.block_writer = BLOCK_WRITERS[codec]
@@ -564,6 +586,19 @@ def writer(fo,
with open('weather.avro', 'wb') as out:
writer(out, parsed_schema, records)
+
+ Given an existing avro file, it's possible to append to it by re-opening
+ the file in `a+b` mode. If the file is only opened in `ab` mode, we aren't
+ able to read some of the existing header information and an error will be
+ raised. For example::
+
+ # Write initial records
+ with open('weather.avro', 'wb') as out:
+ writer(out, parsed_schema, records)
+
+ # Write some more records
+ with open('weather.avro', 'a+b') as out:
+ writer(out, parsed_schema, more_records)
"""
# Sanity check that records is not a single dictionary (as that is a common
# mistake and the exception that gets raised is not helpful)

MODIFY
fastavro/six.py
fastavro/six.py
@@ -45,6 +45,18 @@ if sys.version_info >= (3, 0):
def py3_fstint(datum):
return datum[0]
+ def py3_appendable(file_like):
+ if file_like.tell() != 0:
+ if file_like.readable():
+ return True
+ else:
+ raise ValueError(
+ "When appending to an avro file you must use the " +
+ "'a+' mode, not just 'a'"
+ )
+ else:
+ return False
+
else: # Python 2x
from cStringIO import StringIO as MemoryIO # flake8: noqa
xrange = xrange
@@ -84,6 +96,37 @@ else: # Python 2x
def py2_fstint(datum):
return unpack('!b', datum[0])[0]
+ def _readable(file_like):
+ try:
+ file_like.read()
+ except Exception:
+ return False
+ return True
+
+ def py2_appendable(file_like):
+ # On Python 2 things are a mess. We basically just rely on looking at
+ # the mode. If that doesn't exist (like in the case of an io.BytesIO)
+ # then we check the position and readablility.
+ try:
+ mode = file_like.mode
+ except AttributeError:
+ # This is probably some io stream so we rely on its tell() working
+ if file_like.tell() != 0 and _readable(file_like):
+ return True
+ else:
+ return False
+
+ if "a" in file_like.mode:
+ if "+" in file_like.mode:
+ return True
+ else:
+ raise ValueError(
+ "When appending to an avro file you must use the " +
+ "'a+' mode, not just 'a'"
+ )
+ else:
+ return False
+
# We do it this way and not just redifine function since Cython do not like it
if sys.version_info >= (3, 0):
btou = py3_btou
@@ -97,6 +140,7 @@ if sys.version_info >= (3, 0):
mk_bits = py3_mk_bits
str2ints = py3_bytes2ints
fstint = py3_fstint
+ appendable = py3_appendable
else:
btou = py2_btou
utob = py2_utob
@@ -109,3 +153,4 @@ else:
mk_bits = py2_mk_bits
str2ints = py2_str2ints
fstint = py2_fstint
+ appendable = py2_appendable

MODIFY
tests/test_fastavro.py
tests/test_fastavro.py
@@ -1606,3 +1606,120 @@ def test_union_schema_ignores_extra_fields():
records = [{"name": "abc", "other": "asd"}]
assert [{"name": "abc"}] == roundtrip(schema, records)
+
+
+def test_appending_records(tmpdir):
+ """https://github.com/fastavro/fastavro/issues/276"""
+ schema = {
+ "type": "record",
+ "name": "test_appending_records",
+ "fields": [{
+ "name": "field",
+ "type": "string",
+ }]
+ }
+
+ test_file = str(tmpdir.join("test.avro"))
+
+ with open(test_file, "wb") as new_file:
+ fastavro.writer(new_file, schema, [{"field": "foo"}])
+
+ with open(test_file, "a+b") as new_file:
+ fastavro.writer(new_file, schema, [{"field": "bar"}])
+
+ with open(test_file, "rb") as new_file:
+ reader = fastavro.reader(new_file)
+ new_records = list(reader)
+
+ assert new_records == [{"field": "foo"}, {"field": "bar"}]
+
+
+def test_appending_records_with_io_stream():
+ """https://github.com/fastavro/fastavro/issues/276"""
+ schema = {
+ "type": "record",
+ "name": "test_appending_records_with_io_stream",
+ "fields": [{
+ "name": "field",
+ "type": "string",
+ }]
+ }
+
+ stream = MemoryIO()
+
+ fastavro.writer(stream, schema, [{"field": "foo"}])
+
+ # Should be able to append to the existing stream
+ fastavro.writer(stream, schema, [{"field": "bar"}])
+
+ stream.seek(0)
+ reader = fastavro.reader(stream)
+ new_records = list(reader)
+
+ assert new_records == [{"field": "foo"}, {"field": "bar"}]
+
+ # If we seek to the beginning and write, it will be treated like a brand
+ # new file
+ stream.seek(0)
+ fastavro.writer(stream, schema, [{"field": "abcdefghijklmnopqrstuvwxyz"}])
+
+ stream.seek(0)
+ reader = fastavro.reader(stream)
+ new_records = list(reader)
+
+ assert new_records == [{"field": "abcdefghijklmnopqrstuvwxyz"}]
+
+
+def test_appending_records_wrong_mode_fails(tmpdir):
+ """https://github.com/fastavro/fastavro/issues/276"""
+ schema = {
+ "type": "record",
+ "name": "test_appending_records_wrong_mode_fails",
+ "fields": [{
+ "name": "field",
+ "type": "string",
+ }]
+ }
+
+ test_file = str(tmpdir.join("test.avro"))
+
+ with open(test_file, "wb") as new_file:
+ fastavro.writer(new_file, schema, [{"field": "foo"}])
+
+ with open(test_file, "ab") as new_file:
+ with pytest.raises(ValueError) as exc:
+ fastavro.writer(new_file, schema, [{"field": "bar"}])
+
+ assert "you must use the 'a+' mode, not just 'a'" in str(exc)
+
+
+def test_appending_records_different_schema_fails(tmpdir):
+ """https://github.com/fastavro/fastavro/issues/276"""
+ schema = {
+ "type": "record",
+ "name": "test_appending_records_different_schema_fails",
+ "fields": [{
+ "name": "field",
+ "type": "string",
+ }]
+ }
+
+ test_file = str(tmpdir.join("test.avro"))
+
+ with open(test_file, "wb") as new_file:
+ fastavro.writer(new_file, schema, [{"field": "foo"}])
+
+ different_schema = {
+ "type": "record",
+ "name": "test_appending_records",
+ "fields": [{
+ "name": "field",
+ "type": "int",
+ }]
+ }
+
+ with open(test_file, "a+b") as new_file:
+ with pytest.raises(ValueError) as exc:
+ fastavro.writer(new_file, different_schema, [{"field": 1}])
+
+ assert "does not match file writer_schema" in str(exc)

add support for appending to files