fastavro in-memory container round-trip with logical types: working versions (tested 2026-10-18)

fastavro==1.13.1
pytest==9.1.1

import io
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Any, Iterable

import fastavro

USER_SCHEMA: dict[str, Any] = {
    "type": "record",
    "name": "User",
    "namespace": "example",
    "fields": [
        {"name": "id", "type": "string"},
        {"name": "created", "type": {"type": "long", "logicalType": "timestamp-millis"}},
        {"name": "birthday", "type": {"type": "int", "logicalType": "date"}},
        {
            "name": "balance",
            "type": {"type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 2},
        },
        {"name": "uid", "type": {"type": "string", "logicalType": "uuid"}},
    ],
}

READER_SCHEMA: dict[str, Any] = {
    "type": "record",
    "name": "User",
    "namespace": "example",
    "fields": [
        {"name": "id", "type": "string"},
        {
            "name": "created_at",
            "type": {"type": "long", "logicalType": "timestamp-millis"},
            "aliases": ["created"],
        },
        {
            "name": "balance",
            "type": {"type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 2},
        },
        {"name": "uid", "type": {"type": "string", "logicalType": "uuid"}},
        {"name": "nickname", "type": "string", "default": "anon"},
        {"name": "tier", "type": "string", "default": "free"},
    ],
}

BAD_READER_SCHEMA: dict[str, Any] = {
    "type": "record",
    "name": "User",
    "namespace": "example",
    "fields": [
        {"name": "id", "type": "string"},
        {"name": "nickname", "type": "string"},
    ],
}

USER = fastavro.parse_schema(USER_SCHEMA)
READER = fastavro.parse_schema(READER_SCHEMA)
BAD_READER = fastavro.parse_schema(BAD_READER_SCHEMA)

SAMPLE: dict[str, Any] = {
    "id": "u-1",
    "created": datetime(2021, 2, 3, 4, 5, 6, tzinfo=timezone.utc),
    "birthday": date(1990, 5, 4),
    "balance": Decimal("10.50"),
    "uid": "12345678-1234-5678-1234-567812345678",
}

def write_container(records: Iterable[dict], schema: dict = USER, codec: str = "null") -> bytes:
    buf = io.BytesIO()
    fastavro.writer(buf, schema, list(records), codec=codec)
    return buf.getvalue()

def read_container(data: bytes, reader_schema: dict | None = None) -> list[dict]:
    return list(fastavro.reader(io.BytesIO(data), reader_schema=reader_schema))

def write_one(record: dict, schema: dict = USER) -> bytes:
    buf = io.BytesIO()
    fastavro.schemaless_writer(buf, schema, record)
    return buf.getvalue()

def read_one(data: bytes, schema: dict = USER) -> dict:
    return fastavro.schemaless_reader(io.BytesIO(data), schema)

import io
from datetime import date, datetime
from decimal import Decimal
from uuid import UUID

import fastavro
import pytest
from fastavro.read import SchemaResolutionError

from avro_store import (
    BAD_READER,
    READER,
    SAMPLE,
    read_container,
    read_one,
    write_container,
    write_one,
)

def test_roundtrip_decodes_logical_types():
    (row,) = read_container(write_container([SAMPLE]))
    assert isinstance(row["created"], datetime) and row["created"].tzinfo is not None
    assert row["created"] == SAMPLE["created"]
    assert row["birthday"] == date(1990, 5, 4)
    assert row["balance"] == Decimal("10.50")
    assert row["uid"] == UUID(SAMPLE["uid"])

def test_deflate_codec_is_recorded_in_header():
    data = write_container([SAMPLE], codec="deflate")
    reader = fastavro.reader(io.BytesIO(data))
    assert reader.metadata["avro.codec"] == "deflate"
    assert [row["id"] for row in reader] == ["u-1"]

def test_reader_schema_renames_drops_and_defaults():
    (row,) = read_container(write_container([SAMPLE]), reader_schema=READER)
    assert row["created_at"] == SAMPLE["created"]
    assert row["nickname"] == "anon" and row["tier"] == "free"
    assert "birthday" not in row

def test_new_reader_field_without_default_raises():
    with pytest.raises(SchemaResolutionError):
        read_container(write_container([SAMPLE]), reader_schema=BAD_READER)

def test_schemaless_blob_needs_its_schema():
    raw = write_one(SAMPLE)
    assert raw[:4] != b"Obj\x01"
    with pytest.raises(ValueError):
        read_container(raw)
    assert read_one(raw)["id"] == "u-1"

def test_decimal_scale_beyond_schema_raises_but_shorter_is_padded():
    with pytest.raises(ValueError):
        write_container([dict(SAMPLE, balance=Decimal("10.555"))])
    (row,) = read_container(write_container([dict(SAMPLE, balance=Decimal("10.5"))]))
    assert row["balance"] == Decimal("10.50")

[pytest]
pythonpath = .