Я использую fast avro для десериализации сообщения в нескольких схемах. У меня есть n схем, созданных с помощью pydantic avro. Когда я получаю сообщение, я не знаю точно, в какую схему оно сериализовано, поэтому я пытаюсь десериализовать их в цикле, ожидая десериализации при первом успешном сообщении. Однако иногда fastavro десериализует сообщение по неправильной схеме.
def _deserialize_first(data, schemas):
for schema in schemas:
buffer = BytesIO(data)
try:
return schemaless_reader(buffer, schema)
except Exception:
continue
return None
Например, у меня есть схемы:
class FoundCityMessage(AvroBase):
id: int
action: str
city_name: str
class TapMessage(AvroBase):
id: int
action: str
count_of_taps: int
timestamp: float
Для запроса, сериализованного TapMessage, найденная строка всегда является Tap, а для действия Found = Found. Если я отправлю запрос с действием найдено и city_name = BBB, сообщение десериализуется правильно, но если строка длиннее трех символов, я получаю action = Found, но тогда я не получаю имя города, я получаю count_of_taps и метку времени.
как сделать так, чтобы сообщение десериализовалось только в правильной схеме avro?
Редактировать после предложения из комментариев. Теперь сообщение не десериализуется ни по одной из схем
def _deserialize_first(data, schemas: list[dict]):
for schema in schemas:
buffer = BytesIO(data)
buffer.seek(0)
try:
validate(data, schema, strict=True)
return schemaless_reader(buffer, schema)
except ValidationError:
continue
return None
Подробнее здесь: https://stackoverflow.com/questions/787 ... -correctly
