FASTAVRO(1)

📦
version 1.13.1 · 8 versions
, in fedora 43

💅 FASTAVRO — the manual page, bestie edition ✨

okay so you opened FASTAVRO(1), the manual for fastavro, aka fastavro Documentation, and honestly? you're about to have a great time. buckle up 🎀

🌸 NAME 🌸

fastavro — fastavro Documentation

so here's the tea, bestie: the current Python avro package is dog slow 😭

like, on a test case of about 10K records, it takes about 14sec to iterate over all of them. meanwhile the JAVA avro SDK does the exact same thing in about 1.9sec. the disrespect. 💀

fastavro is an alternative implementation that is much faster. it iterates over the same 10K records in 2.9sec, and if you use it with PyPy it'll do it in 1.5sec (to be fair, the JAVA benchmark is doing some extra JSON encoding/decoding, so we're not being shady, we're being honest 🫶).

and if the optional C extension (generated by Cython) is available, then fastavro will be even faster — for the same 10K records it'll run in about 1.7sec. periodt. 👑

✨ SUPPORTED FEATURES ✨

the way that fastavro just does all of this? iconic. no cap:

  • File Writer 📝
  • File Reader (iterating via records or blocks) 📖
  • Schemaless Writer ✍️
  • Schemaless Reader 👀
  • JSON Writer 🌈
  • JSON Reader 🌈
  • Codecs (Snappy, Deflate, Zstandard, Bzip2, LZ4, XZ) 🗜️
  • Schema resolution 🧩
  • Aliases 🏷️
  • Logical Types 🧠
  • Parsing schemas into the canonical form 📐
  • Schema fingerprinting 🔍

💀 MISSING FEATURES 💀

ngl there's exactly one thing on this list and it's a whole vibe:

  • Anything involving Avro's RPC features

that's it. that's the list. we don't do RPC here, bestie, and we're not sorry about it 🙈

🧚 EXAMPLE 🧚

here's the main character energy you came for — actual code, literally copy-pasteable:

from fastavro import writer, reader, parse_schema

schema = {
    'doc': 'A weather reading.',
    'name': 'Weather',
    'namespace': 'test',
    'type': 'record',
    'fields': [
        {'name': 'station', 'type': 'string'},
        {'name': 'time', 'type': 'long'},
        {'name': 'temp', 'type': 'int'},
    ],
}
parsed_schema = parse_schema(schema)

# 'records' can be an iterable (including generator)
records = [
    {u'station': u'011990-99999', u'temp': 0, u'time': 1433269388},
    {u'station': u'011990-99999', u'temp': 22, u'time': 1433270389},
    {u'station': u'011990-99999', u'temp': -11, u'time': 1433273379},
    {u'station': u'012650-99999', u'temp': 111, u'time': 1433275478},
]

# Writing
with open('weather.avro', 'wb') as out:
    writer(out, parsed_schema, records)

# Reading
with open('weather.avro', 'rb') as fo:
    for record in reader(fo):
        print(record)

quick lil note bestie: 'records' can be an iterable (including a generator) — so you don't have to materialize the whole thing, you can just stream it through. slay. 💅

📚 DOCUMENTATION 📚

(stay tuned, there's more of this page coming — we're not done yet 🦋)

💖 FASTAVRO.READ 💖

Okay so this is the part where we actually read the avro files, and honestly? It's giving main character energy. Let's get into it ✨

🌸 reader

class reader(fo: IO | AvroJSONDecoder, reader_schema: str | List[Any] |
Dict[Any, Any] | None = None, return_record_name: bool = False,
return_record_name_override: bool = False, handle_unicode_errors: str =
'strict', return_named_type: bool = False, return_named_type_override:
bool = False)

Iterator over records in an avro file. Yes bestie, it just iterates, that's the whole vibe 🦋

🎀 Parameters

  • fo — File-like object to read from
  • reader_schema — Reader schema
  • return_record_name — If true, when reading a union of records, the result will be a tuple where the first value is the name of the record and the second value is the record itself
  • return_record_name_override — If true, this will modify the behavior of return_record_name so that the record name is only returned for unions where there is more than one record. For unions that only have one record, this option will make it so that the record is returned by itself, not a tuple with the name.
  • return_named_type — If true, when reading a union of named types, the result will be a tuple where the first value is the name of the type and the second value is the record itself NOTE: Using this option will ignore return_record_name and return_record_name_override
  • return_named_type_override — If true, this will modify the behavior of return_named_type so that the named type is only returned for unions where there is more than one named type. For unions that only have one named type, this option will make it so that the named type is returned by itself, not a tuple with the name
  • handle_unicode_errors — Default strict. Should be set to a valid string that can be used in the errors argument of the string decode() function. Examples include replace and ignore

✨ Example:

from fastavro import reader
with open('some-file.avro', 'rb') as fo:
    avro_reader = reader(fo)
    for record in avro_reader:
        process_record(record)

The fo argument is a file-like object so another common example usage would use an io.BytesIO object like so:

from io import BytesIO
from fastavro import writer, reader

fo = BytesIO()
writer(fo, schema, records)
fo.seek(0)
for record in reader(fo):
    process_record(record)

👑 Attributes you can peek at

  • metadata — Key-value pairs in the header metadata
  • codec — The codec used when writing
  • writer_schema — The schema used when writing
  • reader_schema — The schema used when reading (if provided)

🧚 block_reader

class block_reader(fo: IO, reader_schema: str | List[Any] | Dict[Any,
Any] | None = None, return_record_name: bool = False,
return_record_name_override: bool = False, handle_unicode_errors: str =
'strict', return_named_type: bool = False, return_named_type_override:
bool = False)

Iterator over Block in an avro file. Same energy as reader, just chunked differently 🍓

💅 Parameters

  • fo — Input stream
  • reader_schema — Reader schema
  • return_record_name — If true, when reading a union of records, the result will be a tuple where the first value is the name of the record and the second value is the record itself
  • return_record_name_override — If true, this will modify the behavior of return_record_name so that the record name is only returned for unions where there is more than one record. For unions that only have one record, this option will make it so that the record is returned by itself, not a tuple with the name.
  • return_named_type — If true, when reading a union of named types, the result will be a tuple where the first value is the name of the type and the second value is the record itself NOTE: Using this option will ignore return_record_name and return_record_name_override
  • return_named_type_override — If true, this will modify the behavior of return_named_type so that the named type is only returned for unions where there is more than one named type. For unions that only have one named type, this option will make it so that the named type is returned by itself, not a tuple with the name
  • handle_unicode_errors — Default strict. Should be set to a valid string that can be used in the errors argument of the string decode() function. Examples include replace and ignore

🌈 Example:

from fastavro import block_reader
with open('some-file.avro', 'rb') as fo:
    avro_reader = block_reader(fo)
    for block in avro_reader:
        process_block(block)

💜 Attributes you can peek at

  • metadata — Key-value pairs in the header metadata
  • codec — The codec used when writing
  • writer_schema — The schema used when writing
  • reader_schema — The schema used when reading (if provided)

🦋 Block

class Block(bytes_, num_records, codec, reader_schema, writer_schema,
named_schemas, offset, size, options)

An avro block. Will yield records when iterated over 🔥

  • num_records — Number of records in the block
  • writer_schema — The schema used when writing
  • reader_schema — The schema used when reading (if provided)
  • offset — Offset of the block from the beginning of the avro file
  • size — Size of the block in bytes

🌟 schemaless_reader

schemaless_reader(fo: IO, writer_schema: str | List[Any] | Dict[Any,
Any], reader_schema: str | List[Any] | Dict[Any, Any] | None = None,
return_record_name: bool = False, return_record_name_override: bool =
False, handle_unicode_errors: str = 'strict', return_named_type: bool =
False, return_named_type_override: bool = False) -> None | str | float |
int | Decimal | bool | bytes | List[Any] | Dict[Any, Any]

Reads a single record written using the schemaless_writer() ✨

🎀 Parameters

  • fo — Input stream
  • writer_schema — Schema used when calling schemaless_writer
  • reader_schema — If the schema has changed since being written then the new schema can be given to allow for schema migration
  • return_record_name — If true, when reading a union of records, the result will be a tuple where the first value is the name of the record and the second value is the record itself
  • return_record_name_override — If true, this will modify the behavior of return_record_name so that the record name is only returned for unions where there is more than one record. For unions that only have one record, this option will make it so that the record is returned by itself, not a tuple with the name.
  • return_named_type — If true, when reading a union of named types, the result will be a tuple where the first value is the name of the type and the second value is the record itself NOTE: Using this option will ignore return_record_name and return_record_name_override
  • return_named_type_override — If true, this will modify the behavior of return_named_type so that the named type is only returned for unions where there is more than one named type. For unions that only have one named type, this option will make it so that the named type is returned by itself, not a tuple with the name
  • handle_unicode_errors — Default strict. Should be set to a valid string that can be used in the errors argument of the string decode() function. Examples include replace and ignore

💖 Example:

parsed_schema = fastavro.parse_schema(schema)
with open('file', 'rb') as fp:
    record = fastavro.schemaless_reader(fp, parsed_schema)
Note: The schemaless_reader can only read a single record. One (1). Singular. Don't come for her 💀

🌸 is_avro

is_avro(path_or_buffer: str | IO) -> bool

Return True if path (or buffer) points to an Avro file. This will only work for avro files that contain the normal avro schema header like those create from writer(). This function is not intended to be used with binary data created from schemaless_writer() since that does not include the avro header. No cap 🫶

🌈 Parameters

  • path_or_buffer — Path to file

✨ FASTAVRO.WRITE ✨

writer(fo: IO | AvroJSONEncoder, schema: str | List[Any] | Dict[Any,
Any], records: Iterable[Any], codec: str = 'null', sync_interval: int =
16000, metadata: Dict[str, str] | None = None, validator: bool = False,
sync_marker: bytes = b'', codec_compression_level: int | None = None, *,
strict: bool = False, strict_allow_default: bool = False,
disable_tuple_notation: bool = False)

okay bestie this is the one 💖 it writes your records out to fo (which is a stream, obvi) following whatever schema you hand it. this is literally the whole point of the library so pay attention ✨

Parameters

  • fo — the output stream, your lil destination for all this data 💅
  • schema — the writer schema, the blueprint, the moment 👑
  • records — the records you wanna write. usually this is a list of dictionary representations of the records, but lowkey it can be any iterable, she's flexible like that
  • codec — compression codec, and your options are 'null', 'deflate', or 'snappy' (that last one only if it's installed, no cap)
  • sync_interval — the size of the sync interval
  • metadata — header metadata
  • validator — if true, validation gets done on the records
  • sync_marker — a byte string used as the avro sync marker. if you don't provide one, a random byte string gets used instead, so it's giving surprise me ✨
  • codec_compression_level — compression level to use with the specified codec (if the codec supports it, so check first bestie)
  • strict — if you set this to True, an error will be raised if records don't contain exactly the same fields that the schema states. she's strict, she said what she said 💅
  • strict_allow_default — if set to True, an error will be raised if records don't contain exactly the same fields that the schema states, unless it's a missing field that has a default value in the schema. so it's strict but with mercy, iconic
  • disable_tuple_notation — if set to True, tuples won't be treated as a special case. therefore, using a tuple to indicate the type of a record will not work 🫠

Example:

from fastavro import writer, parse_schema

schema = {
    'doc': 'A weather reading.',
    'name': 'Weather',
    'namespace': 'test',
    'type': 'record',
    'fields': [
        {'name': 'station', 'type': 'string'},
        {'name': 'time', 'type': 'long'},
        {'name': 'temp', 'type': 'int'},
    ],
}
parsed_schema = parse_schema(schema)

records = [
    {u'station': u'011990-99999', u'temp': 0, u'time': 1433269388},
    {u'station': u'011990-99999', u'temp': 22, u'time': 1433270389},
    {u'station': u'011990-99999', u'temp': -11, u'time': 1433273379},
    {u'station': u'012650-99999', u'temp': 111, u'time': 1433275478},
]

with open('weather.avro', 'wb') as out:
    writer(out, parsed_schema, records)

the fo argument is a file-like object, so another common usage is passing an io.BytesIO object instead, like so 🎀

from io import BytesIO
from fastavro import writer

fo = BytesIO()
writer(fo, schema, records)

now here's the tea 🍵 if you've got an existing avro file, you can append to it by re-opening the file in a+b mode. but 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. so a+b or nothing, bestie 💀

# 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, None, more_records)
Note: when appending, any schema provided will be ignored since the schema in the avro file will be re-used. therefore it's convenient to just use None as the schema. periodt 💅

schemaless_writer

schemaless_writer(fo: IO, schema: str | List[Any] | Dict[Any, Any],
record: Any, *, strict: bool = False, strict_allow_default: bool =
False, disable_tuple_notation: bool = False)

writes a single record without the schema or header information. lowkey it's the stripped-down version, no frills, just vibes ✨

Parameters

  • fo — output file
  • schema — the schema
  • record — the record to write
  • strict — if set to True, an error will be raised if records don't contain exactly the same fields that the schema states
  • strict_allow_default — if set to True, an error will be raised if records don't contain exactly the same fields that the schema states, unless it's a missing field that has a default value in the schema
  • disable_tuple_notation — if set to True, tuples won't be treated as a special case. therefore, using a tuple to indicate the type of a record will not work 🫠

Example:

parsed_schema = fastavro.parse_schema(schema)
with open('file', 'wb') as fp:
    fastavro.schemaless_writer(fp, parsed_schema, record)
Note: the schemaless_writer can only write a single record. one and done, no more, she's not doing a series 🎀

Using the tuple notation to specify which branch of a union to take

okay so since this library uses plain dictionaries to represent a record, it's totally possible for that dictionary to fit the definition of two different records. chaos, honestly 😭

for example, given a dictionary like this:

{"name": "My Name"}

it would be valid against both of these records:

child_schema = {
    "name": "Child",
    "type": "record",
    "fields": [
        {"name": "name", "type": "string"},
        {"name": "favorite_color", "type": ["null", "string"]},
    ]
}

pet_schema = {
    "name": "Pet",
    "type": "record",
    "fields": [
        {"name": "name", "type": "string"},
        {"name": "favorite_toy", "type": ["null", "string"]},
    ]
}

this becomes a whole problem when a schema contains a union of these two similar records, because it's not clear which record the dictionary represents. like bestie, which one 😩 for example, if you used the previous dictionary with the following schema, it wouldn't be clear if the record should be serialized as a Child or a Pet:

household_schema = {
    "name": "Household",
    "type": "record",
    "fields": [
        {"name": "address", "type": "string"},
        {
            "name": "family_members",
            "type": {
                "type": "array", "items": [
                    {
                        "name": "Child",
                        "type": "record",
                        "fields": [
                            {"name": "name", "type": "string"},
                            {"name": "favorite_color", "type": ["null", "string"]},
                        ]
                    }, {
                        "name": "Pet",
                        "type": "record",
                        "fields": [
                            {"name": "name", "type": "string"},
                            {"name": "favorite_toy", "type": ["null", "string"]},
                        ]
                    }
                ]
            }
        },
    ]
}

to resolve this, you can use a tuple notation where the first value of the tuple is the fully namespaced record name and the second value is the dictionary. slay, problem solved 💖 for example:

records = [
    {
        "address": "123 Drive Street",
        "family_members": [
            ("Child", {"name": "Son"}),
            ("Child", {"name": "Daughter"}),
            ("Pet", {"name": "Dog"}),
        ]
    }
]

Using the record hint to specify which branch of a union to take

in addition to the tuple notation for specifying the name of a record, you can also include a special -type attribute (note that this attribute is -type, not type, ok bestie read it again 👀) on a record to do the same thing. so the example above which looked like this:

records = [
    {
        "address": "123 Drive Street",
        "family_members": [
            ("Child", {"name": "Son"}),
            ("Child", {"name": "Daughter"}),
            ("Pet", {"name": "Dog"}),
        ]
    }
]

would now look like this:

records = [
    {
        "address": "123 Drive Street",
        "family_members": [
            {"-type": "Child", "name": "Son"},
            {"-type": "Child", "name": "Daughter"},
            {"-type": "Pet", "name": "Dog"},
        ]
    }
]

unlike the tuple notation, which can be used with any avro type in a union, this -type hint can only be used with records. however, this can be useful if you want to make a single record dictionary that can be used both in and out of unions. versatile queen 👑

🌈 FASTAVRO.JSON_READ 🌈

json_reader

json_reader(fo: ~typing.IO, schema: str | ~typing.List[~typing.Any] |
~typing.Dict[~typing.Any, ~typing.Any], reader_schema: str |
~typing.List[~typing.Any] | ~typing.Dict[~typing.Any, ~typing.Any] |
None = None, *, decoder=<class
'fastavro.io.json_decoder.AvroJSONDecoder'>) -> reader

an iterator over records in an avro json file ✨ she just keeps handing you records one by one, generous behavior honestly

Parameters

  • fo — file-like object to read from
  • schema — the original schema used when writing the JSON data
  • reader_schema — if the schema has changed since being written, then the new schema can be given to allow for schema migration. glow up moment 💅
  • decoder — by default the standard AvroJSONDecoder will be used, but a custom one could be passed here. iykyk

Example:

from fastavro import json_reader

schema = {
    'doc': 'A weather reading.',
    'name': 'Weather',
    'namespace': 'test',
    'type': 'record',
    'fields': [
        {'name': 'station', 'type': 'string'},
        {'name': 'time', 'type': 'long'},
        {'name': 'temp', 'type': 'int'},
    ]
}

with open('some-file', 'r') as fo:
    avro_reader = json_reader(fo, schema)
    for record in avro_reader:
        print(record)

✨ FASTAVRO.JSON_WRITE ✨

okay bestie, meet the write-side twin of json_read — it's giving main character energy fr 💖

json_writer(fo: ~typing.IO, schema: str | ~typing.List[~typing.Any] |
~typing.Dict[~typing.Any, ~typing.Any], records:
~typing.Iterable[~typing.Any], *, write_union_type: bool = True,
validator: bool = False, encoder=<class
'fastavro.io.json_encoder.AvroJSONEncoder'>, strict: bool = False,
strict_allow_default: bool = False, disable_tuple_notation: bool =
False) -> None

Write records to fo (stream) according to schema. that's the whole vibe, but the details are where the magic lives ✨

🎀 Parameters

  • fo — File-like object to write to. like, a stream, a file, whatever speaks the write() language, she doesn't judge 🫶
  • schema — Writer schema. the blueprint, the rules of the universe for your records
  • records — Records to write. This is commonly a list of the dictionary representation of the records, but it can be any iterable. yes bestie, literally any iterable, she's flexible like that 🌸
  • write_union_type — Determine whether to write the union type in the json message. If this is set to False the output will be clear json. It may however not be decodable back to avro record by json_read. ⚠️ heads up: turn this off and you get pretty clean json, but you might not be able to round-trip it back. choices were made. 💀
  • validator — If true, validation will be done on the records. she's checking your work, no skipping steps 🔥
  • encoder — By default the standard AvroJSONEncoder will be used, but a custom one could be passed here. you can bring your own encoder to the party, iconic 💅
  • strict — If set to True, an error will be raised if records do not contain exactly the same fields that the schema states. exactly the same fields, no extras, no missing, no cap 🚨
  • strict_allow_default — If set to True, an error will be raised if records do not contain exactly the same fields that the schema states unless it is a missing field that has a default value in the schema. so it's strict but with a little grace for the fields that already have a fallback 🍓
  • strict_allow_default ... wait no bestie, that's the same one — moving on 😭
  • disable_tuple_notation — If set to True, tuples will not be treated as a special case. Therefore, using a tuple to indicate the type of a record will not work. turn this on and your tuple tricks just... stop working. she said no more 🙈

🌈 Example

look at her go, she's perfect 💖

from fastavro import json_writer, parse_schema

schema = {
    'doc': 'A weather reading.',
    'name': 'Weather',
    'namespace': 'test',
    'type': 'record',
    'fields': [
        {'name': 'station', 'type': 'string'},
        {'name': 'time', 'type': 'long'},
        {'name': 'temp', 'type': 'int'},
    ],
}
parsed_schema = parse_schema(schema)

records = [
    {u'station': u'011990-99999', u'temp': 0, u'time': 1433269388},
    {u'station': u'011990-99999', u'temp': 22, u'time': 1433270389},
    {u'station': u'011990-99999', u'temp': -11, u'time': 1433273379},
    {u'station': u'012650-99999', u'temp': 111, u'time': 1433275478},
]

with open('some-file', 'w') as out:
    json_writer(out, parsed_schema, records)

notice how she parses the schema first with parse_schema, opens a file in 'w' mode, and then just... sends it. clean. effortless. queen behavior. 👑

✨ FASTAVRO.SCHEMA ✨

🎀 parse_schema

parse_schema(schema: str | List[Any] | Dict[Any, Any], named_schemas: Dict[str, Dict[Any, Any]] | None = None, *, expand: bool = False, _write_hint: bool = True, _force: bool = False, _ignore_default_error: bool = False) -> str | List[Any] | Dict[Any, Any]

okay so bestie, this one returns a parsed avro schema. 💖

ngl you don't have to call parse_schema, but if you do and you save the parsed schema to reuse later, all your future operations will be faster because the schema won't need to get reparsed over and over. it's giving efficiency queen. 👑

Parameters 🌸

  • schema — the input schema, bestie
  • named_schemas — dictionary mapping named schemas to their schema definition
  • expand — if true, named schemas get fully expanded to their true schemas instead of just being represented by their name. ⚠️ lowkey important: this format is meant to be output only, so do NOT pass it into other reader/writer functions — it doesn't conform to the avro specification and will probably throw an exception at you. 💀
  • _write_hint — internal API argument that says whether or not the __fastavro_parsed marker should be added to the schema
  • _force — internal API argument. if True, the schema gets parsed anyway even if it's already been parsed and has the __fastavro_parsed marker. like, it does not care. 💅
  • _ignore_default_error — internal API argument. if True, when a union has the wrong default value, no error gets raised.

Example: ✨

from fastavro import parse_schema
from fastavro import writer

parsed_schema = parse_schema(original_schema)
with open('weather.avro', 'wb') as out:
    writer(out, parsed_schema, records)

so sometimes you might have two schemas where one schema references another. for the sake of example, let's assume you've got a Parent schema that references a Child schema. if you tried to parse the parent schema all by itself, you'd get an exception because the child schema isn't defined yet. 😭 to fix this, we can use the named_schemas argument to pass a shared dictionary when parsing both of the schemas. the dictionary gets populated with the necessary schema references to make parsing possible. fr fr. like this:

from fastavro import parse_schema

named_schemas = {}
parsed_child = parse_schema(child_schema, named_schemas)
parsed_parent = parse_schema(parent_schema, named_schemas)

🦋 fullname

fullname(schema: Dict[Any, Any]) -> str

Returns the fullname of a schema. periodt. ✨

Parameters

  • schema — the input schema, hun

Example:

from fastavro.schema import fullname

schema = {
    'doc': 'A weather reading.',
    'name': 'Weather',
    'namespace': 'test',
    'type': 'record',
    'fields': [
        {'name': 'station', 'type': 'string'},
        {'name': 'time', 'type': 'long'},
        {'name': 'temp', 'type': 'int'},
    ],
}

fname = fullname(schema)
assert fname == "test.Weather"

🌈 expand_schema

expand_schema(schema: str | List[Any] | Dict[Any, Any]) -> str | List[Any] | Dict[Any, Any]

Returns a schema where all named types are expanded to their real schema. the way this one un-collapses everything… iconic. 💅

NOTE: the output of this function produces a schema that can include multiple definitions of the same named type (as per design) which are not valid per the avro specification. so do NOT pass the output of this to the normal writer/reader functions — it will likely result in an error. consider yourself warned bestie. 😭

Parameters

  • schema (dict) — the input schema

Example:

from fastavro.schema import expand_schema

original_schema = {
    "name": "MasterSchema",
    "namespace": "com.namespace.master",
    "type": "record",
    "fields": [{
        "name": "field_1",
        "type": {
            "name": "Dependency",
            "namespace": "com.namespace.dependencies",
            "type": "record",
            "fields": [
                {"name": "sub_field_1", "type": "string"}
            ]
        }
    }, {
        "name": "field_2",
        "type": "com.namespace.dependencies.Dependency"
    }]
}

expanded_schema = expand_schema(original_schema)

assert expanded_schema == {
    "name": "com.namespace.master.MasterSchema",
    "type": "record",
    "fields": [{
        "name": "field_1",
        "type": {
            "name": "com.namespace.dependencies.Dependency",
            "type": "record",
            "fields": [
                {"name": "sub_field_1", "type": "string"}
            ]
        }
    }, {
        "name": "field_2",
        "type": {
            "name": "com.namespace.dependencies.Dependency",
            "type": "record",
            "fields": [
                {"name": "sub_field_1", "type": "string"}
            ]
        }
    }]
}

🧚 load_schema

load_schema(schema_path: str, *, repo: AbstractSchemaRepository | None = None, named_schemas: Dict[str, Dict[Any, Any]] | None = None, _write_hint: bool = True, _injected_schemas: Set[str] | None = None) -> str | List[Any] | Dict[Any, Any]

Returns a schema loaded from repository. ✨

it will recursively load referenced schemas, attempting to load them from the same repository, using schema_path as the schema name. it's giving chain reaction, but make it schemas. 🦋

if repo isn't provided, FlatDictRepository is used. FlatDictRepository will try to load schemas from the same directory, assuming files are named with the convention <full_name>.avsc.

Parameters 💖

  • schema_path — full schema name, or path to schema file if the default repo is used
  • repo — schema repository instance
  • named_schemas — dictionary of named schemas to their schema definition
  • _write_hint — internal API argument specifying whether or not the __fastavro_parsed marker should be added to the schema
  • _injected_schemas — internal API argument. set of names that have been injected

consider the following example with the default FlatDictRepository… 👑

namespace.Parent.avsc:

{
    "type": "record",
    "name": "Parent",
    "namespace": "namespace",
    "fields": [
        {
            "name": "child",
            "type": "Child"
        }
    ]
}

namespace.Child.avsc:

{
    "type": "record",
    "namespace": "namespace",
    "name": "Child",
    "fields": []
}

Code:

from fastavro.schema import load_schema

parsed_schema = load_schema("namespace.Parent.avsc")

🌸 load_schema_ordered

load_schema_ordered(ordered_schemas: List[str], *, _write_hint: bool = True) -> str | List[Any] | Dict[Any, Any]

Returns a schema loaded from a list of schemas. ✨

the list of schemas should be ordered such that any dependencies are listed before any other schemas that use those dependencies. so if schema A depends on schema B, and schema B depends on schema C, then your list should be [C, B, A]. it's giving ✨correct order or chaos✨ energy.

Parameters

  • ordered_schemas — list of paths to schemas
  • _write_hint — internal API argument specifying whether or not the __fastavro_parsed marker should be added to the schema

consider the following example… 💅

Parent.avsc:

{
    "type": "record",
    "name": "Parent",
    "namespace": "namespace",
    "fields": [
        {
            "name": "child",
            "type": "Child"
        }
    ]
}

namespace.Child.avsc:

{
    "type": "record",
    "namespace": "namespace",
    "name": "Child",
    "fields": []
}

Code:

from fastavro.schema import load_schema_ordered

parsed_schema = load_schema_ordered(
    ["path/to/namespace.Child.avsc", "path/to/Parent.avsc"]
)

💜 to_parsing_canonical_form

to_parsing_canonical_form(schema: str | List[Any] | Dict[Any, Any]) -> str

Returns a string represening the parsing canonical form of the schema. 🎀

for more details on the parsing canonical form, see here: ‐ https://avro.apache.org/docs/current/spec.html#Parsing+Canonical+Form+for+Schemas

Parameters

  • schema — schema to transform

⭐ fingerprint

fingerprint(parsing_canonical_form: str, algorithm: str) -> str

Returns a string represening a fingerprint/hash of the parsing canonical form of a schema. it's giving ✨unique identifier vibes✨ 🔥

for more details on the fingerprint, see here: ‐ https://avro.apache.org/docs/current/spec.html#schema_fingerprints

Parameters

  • parsing_canonical_form — the parsing canonical form of a schema
  • algorithm — the hashing algorithm

✨ FASTAVRO.VALIDATION ✨

okay so like… you have data. you have a schema. are they, like, compatible?? this is the module that tells you 💖

validate(datum: Any, schema: str | List[Any] | Dict[Any, Any], field: str = '', raise_errors: bool = True, strict: bool = False, disable_tuple_notation: bool = False) -> bool

Checks whether a python datum is actually an instance of a schema. vibe check for your records, basically 🫶

Parameters

  • datum — the data you're validating, bestie
  • schema — the schema
  • field — the record field that's being validated
  • raise_errors — if True, invalid data makes errors get raised 😤. if False, you just get a simple True (valid) or False (invalid) and nobody yells
  • strict — if True, fields with no value will raise errors instead of quietly defaulting to None. no silent sneaking 🙈
  • disable_tuple_notation — if you set this to True, tuples stop being treated as a special case. so using a tuple to indicate the type of a record? yeah that won't work anymore 💀

Example:

from fastavro.validation import validate
schema = {...}
record = {...}
validate(record, schema)
validate_many(records: Iterable[Any], schema: str | List[Any] | Dict[Any, Any], raise_errors: bool = True, strict: bool = False, disable_tuple_notation: bool = False) -> bool

Validate a whole list of data!! the group-chat version of validate, fr fr 🌸

Parameters

  • records — the list of records you wanna validate
  • schema — the schema
  • raise_errors — if True, errors get raised for invalid data. if False, you get a plain True (valid) or False (invalid) back
  • strict — if True, fields without values raise errors instead of implicitly defaulting to None
  • disable_tuple_notation — set it to True and tuples are no longer a special case, so using a tuple to indicate the type of a record will not work

Example:

from fastavro.validation import validate_many
schema = {...}
records = [{...}, {...}, ...]
validate_many(records, schema)

🧚 FASTAVRO.UTILS 🧚

fake data generator central!! no cap, this module is so underrated ✨

generate_one(schema: str | List[Any] | Dict[Any, Any]) -> Any

Hands you back a single instance of arbitrary data that conforms to the schema. one (1) little guy, fully schema-legal 👑

Parameters schema — the schema your data should conform to

Example:

from fastavro import schemaless_writer
from fastavro.utils import generate_one

schema = {
    'doc': 'A weather reading.',
    'name': 'Weather',
    'namespace': 'test',
    'type': 'record',
    'fields': [
        {'name': 'station', 'type': 'string'},
        {'name': 'time', 'type': 'long'},
        {'name': 'temp', 'type': 'int'},
    ],
}

with open('weather.avro', 'wb') as out:
    schemaless_writer(out, schema, generate_one(schema))
generate_many(schema: str | List[Any] | Dict[Any, Any], count: int) -> Iterator[Any]

A generator that yields arbitrary data that conforms to the schema. it'll yield a number of data structures equal to whatever you pass in as the count. infinite besties hack (well, count-many besties hack) 💅

Parameters

  • schema — the schema your data should conform to
  • count — how many objects to generate

Example:

from fastavro import writer
from fastavro.utils import generate_many

schema = {
    'doc': 'A weather reading.',
    'name': 'Weather',
    'namespace': 'test',
    'type': 'record',
    'fields': [
        {'name': 'station', 'type': 'string'},
        {'name': 'time', 'type': 'long'},
        {'name': 'temp', 'type': 'int'},
    ],
}

with open('weather.avro', 'wb') as out:
    writer(out, schema, generate_many(schema, 5))
anonymize_schema(schema: str | List[Any] | Dict[Any, Any]) -> str | List[Any] | Dict[Any, Any]

Returns an anonymized schema 🕵️‍♀️✨

Parameters schema — the schema to anonymize

Example:

from fastavro.utils import anonymize_schema

anonymized_schema = anonymize_schema(original_schema)

🌈 LOGICAL TYPES 🌈

okay so logical types are like… the aesthetic layer on top of the raw avro types. same bytes, cuter meaning 💖

Fastavro supports these official logical types:

  • decimal
  • uuid
  • date
  • time-millis
  • time-micros
  • timestamp-millis
  • timestamp-micros
  • local-timestamp-millis
  • local-timestamp-micros

and the official logical type that fastavro is missing support for? 😭

  • duration

🎀 How to specify logical types in your schemas 🎀

So the docs say that when you want something to be a logical type, you just add a logicalType key. Which sounds easy!! but unfortunately this means a super common confusion is people taking a pre-existing schema like this:

schema = {
    "type": "record",
    "name": "root",
    "fields": [
        {
            "name": "id",
            "type": "string",
        },
    ]
}

and then adding the uuid logical type like this:

schema = {
    "type": "record",
    "name": "root",
    "fields": [
        {
            "name": "id",
            "type": "string",
            "logicalType": "uuid",  # This is the wrong place to add this key
        },
    ]
}

But bestie… that adds the logicalType key to the field schema, and that is NOT correct 🚫. What you actually want is to group it with the string like so:

schema = {
    "type": "record",
    "name": "root",
    "fields": [
        {
            "name": "id",
            "type": {
                "type": "string",
                "logicalType": "uuid",  # This is the correct place to add this key
            },
        },
    ]
}

that's it. that's the whole thing. nest it, don't sprinkle it 🙏

🦋 Custom Logical Types 🦋

The Avro specification defines a handful of logical types that most implementations support. For example, one of the defined logical types is a microsecond precision timestamp. The specification states that this value will get encoded as an avro long type.

For the sake of an example, let's say you want to create a new logical type for a microsecond precision timestamp that uses a string as the underlying avro type.

To do this, there are a few functions that need to be defined. First, we need an encoder function that will encode a datetime object as a string. The encoder function is called with two arguments: the data and the schema. So we could define this like so:

def encode_datetime_as_string(data, schema):
    if isinstance(data, datetime):
        return datetime.isoformat(data)
    return data

# or

def encode_datetime_as_string(data, *args):
    if isinstance(data, datetime):
        return datetime.isoformat(data)
    return data
NOTE: You should always check that the data is the expected type in your encoder function before you transform it, and if it is not, then you should return the data unchanged. If you use this custom logical type as a member of a union type, then this encoder function will be called on any value that is provided for a field of that type, even if the value's type is a different member of the union type. 👀 (yes really, every single one)

Then, we need a decoder function that will transform the string back into a datetime object. The decoder function is called with three arguments: the data, the writer schema, and the reader schema. So we could define this like so:

def decode_string_as_datetime(data, writer_schema, reader_schema):
    return datetime.fromisoformat(data)

# or

def decode_string_as_datetime(data, *args):
    return datetime.fromisoformat(data)

Finally, we need to tell fastavro to use these functions. The schema for this custom logical type will use the type string and can use whatever name you would like as the logicalType. In this example, let's suppose we call the logicalType datetime2. To have the library actually use the custom logical type, we use the name of <avro_type>-<logical_type>, so in this example that name would be string-datetime2 and then we add those functions like so:

fastavro.write.LOGICAL_WRITERS["string-datetime2"] = encode_datetime_as_string
fastavro.read.LOGICAL_READERS["string-datetime2"] = decode_string_as_datetime

And you are done. 🎉 Now if the library comes across a schema with a logical type of datetime2 and an avro type of string, it will use the custom functions. For a complete example, see here:

import io
from datetime import datetime

import fastavro
from fastavro import writer, reader

def encode_datetime_as_string(data, *args):
    return datetime.isoformat(data)

def decode_string_as_datetime(data, *args):
    return datetime.fromisoformat(data)

fastavro.write.LOGICAL_WRITERS["string-datetime2"] = encode_datetime_as_string
fastavro.read.LOGICAL_READERS["string-datetime2"] = decode_string_as_datetime

writer_schema = fastavro.parse_schema({
    "type": "record",
    "name": "root",
    "fields": [
        {
            "name": "some_date",
            "type": [
                "null",
                {
                    "type": "string",
                    "logicalType": "datetime2",
                },
            ],
        },
    ]
})

records = [
    {"some_date": datetime.now()}
]

bio = io.BytesIO()

writer(bio, writer_schema, records)

bio.seek(0)

for record in reader(bio):
    print(record)

periodt. custom logical type acquired 💅✨

🫶 FASTAVRO COMMAND LINE SCRIPT 🫶

okay bestie, this is the part where the library stops being just a library and shows up as an actual vibe you can run in your terminal 💖 a command line script gets installed right alongside the library, and its whole job is dumping the contents of your avro file(s) straight to the standard output. no cap.

Usage:

usage: fastavro [-h] [--schema] [--codecs] [--version] [-p] [file [file ...]]

iter over avro file, emit records as JSON

positional arguments:
  file          file(s) to parse

optional arguments:
  -h, --help    show this help message and exit
  --schema      dump schema instead of records
  --codecs      print supported codecs
  --version     show program's version number and exit
  -p, --pretty  pretty print json

so like, breaking that down in bestie terms 🌸

  • file — the file(s) to parse. yes plural, you can pile them on.
  • -h, --help — shows this help message and dips, classic.
  • --schema — dumps the schema instead of the records. it's giving "show me the blueprint".
  • --codecs — prints the supported codecs.
  • --version — shows the program's version number and exits, bye 👋
  • -p, --pretty — pretty print json, because compact json is lowkey unreadable.

✨ Examples ✨

💜 Read an avro file

$ fastavro weather.avro

{"temp": 0, "station": "011990-99999", "time": -619524000000}
{"temp": 22, "station": "011990-99999", "time": -619506000000}
{"temp": -11, "station": "011990-99999", "time": -619484400000}
{"temp": 111, "station": "012650-99999", "time": -655531200000}
{"temp": 78, "station": "012650-99999", "time": -655509600000}

look at those records just streaming out, one per line, main character energy 🔥

💙 Show the schema

$ fastavro --schema weather.avro

{
 "type": "record",
 "namespace": "test",
 "doc": "A weather reading.",
 "fields": [
  {
   "type": "string",
   "name": "station"
  },
  {
   "type": "long",
   "name": "time"
  },
  {
   "type": "int",
   "name": "temp"
  }
 ],
 "name": "Weather"
}

and there's the whole schema, every field laid out, no secrets left bestie 🦋

🌈 FASTAVRO.IO 🌈

this is where the low-level input/output classes live, iykyk.

🎀 fastavro.io.json_decoder 🎀

class AvroJSONDecoder(fo: IO)

Decoder for the avro JSON format. it takes your JSON and turns it back into avro-shaped goodness 💖

NOTE: All attributes and methods on this class should be considered private. do not touch, bestie, it's not for you 🙈

Parameters

  • fo — File-like object to reader from.

🎀 fastavro.io.json_encoder 🎀

class AvroJSONEncoder(fo: IO, *, write_union_type: bool = True)

Encoder for the avro JSON format. this one goes the other way — avro out, JSON in ✨

NOTE: All attributes and methods on this class should be considered private. same energy as above, hands off 💅

Parameters

  • fo — Input stream
  • write_union_type — Determine whether to write the union type in the json message.

🧚 FASTAVRO.REPOSITORY 🧚

🍓 fastavro.repository.base 🍓

class AbstractSchemaRepository

the abstract base situation for schema repositories. that's the whole class, bestie, it's abstract, it's not gonna give you more than that 💀

  • Index
  • Module Index
  • Search Page

👑 AUTHOR 👑

Miki Tebeka

2012, Miki Tebeka

and that's a wrap on the whole page, bestie 🎀 source is FASTAVRO(1), version 1.13.1, dated Oct 08, 2026 — go forth and slay your avro files 💅✨

📦 where this page comes from