add support for local-timestamp micros and millis
MODIFY
docs/logical_types.rst
docs/logical_types.rst
@@ -10,11 +10,11 @@ Fastavro supports the following official logical types:
* time-micros
* timestamp-millis
* timestamp-micros
+* local-timestamp-millis
+* local-timestamp-micros
Fastavro is missing support for the following official logical types:
-* local-timestamp-millis
-* local-timestamp-micros
* duration
How to specify logical types in your schemas

MODIFY
fastavro/_logical_writers.pyx
fastavro/_logical_writers.pyx
@@ -63,6 +63,17 @@ cpdef prepare_timestamp_millis(object data, schema):
return data
+cpdef prepare_local_timestamp_millis(object data, schema):
+ if isinstance(data, datetime.datetime):
+ delta = data.replace(tzinfo=datetime.timezone.utc) - epoch
+ return <long64>(
+ (delta.days * 24 * 3600 + delta.seconds) * MLS_PER_SECOND
+ + int(delta.microseconds / 1000)
+ )
+ else:
+ return data
+
+
cpdef prepare_timestamp_micros(object data, schema):
cdef object tt
cdef tm time_tuple
@@ -96,6 +107,17 @@ cpdef prepare_timestamp_micros(object data, schema):
return data
+cpdef prepare_local_timestamp_micros(object data, schema):
+ if isinstance(data, datetime.datetime):
+ delta = data.replace(tzinfo=datetime.timezone.utc) - epoch
+ return <long64>(
+ (delta.days * 24 * 3600 + delta.seconds) * MCS_PER_SECOND
+ + delta.microseconds
+ )
+ else:
+ return data
+
+
cpdef prepare_date(object data, schema):
if isinstance(data, datetime.date):
return data.toordinal() - const.DAYS_SHIFT
@@ -228,7 +250,9 @@ cpdef prepare_time_micros(object data, schema):
LOGICAL_WRITERS = {
"long-timestamp-millis": prepare_timestamp_millis,
+ "long-local-timestamp-millis": prepare_local_timestamp_millis,
"long-timestamp-micros": prepare_timestamp_micros,
+ "long-local-timestamp-micros": prepare_local_timestamp_micros,
"int-date": prepare_date,
"bytes-decimal": prepare_bytes_decimal,
"fixed-decimal": prepare_fixed_decimal,

MODIFY
fastavro/_logical_writers_py.py
fastavro/_logical_writers_py.py
@@ -5,6 +5,7 @@ import decimal
from io import BytesIO
import os
import time
+from typing import Dict, Union
import uuid
from .const import (
MCS_PER_HOUR,
@@ -46,6 +47,24 @@ def prepare_timestamp_millis(data, schema):
return data
+def prepare_local_timestamp_millis(
+ data: Union[datetime.datetime, int], schema: Dict
+) -> int:
+ """Converts datetime.datetime object to int timestamp with milliseconds.
+
+ The local-timestamp-millis logical type represents a timestamp in a local
+ timezone, regardless of what specific time zone is considered local, with a
+ precision of one millisecond.
+ """
+ if isinstance(data, datetime.datetime):
+ delta = data.replace(tzinfo=datetime.timezone.utc) - epoch
+ return (delta.days * 24 * 3600 + delta.seconds) * MLS_PER_SECOND + int(
+ delta.microseconds / 1000
+ )
+ else:
+ return data
+
+
def prepare_timestamp_micros(data, schema):
"""Converts datetime.datetime to int timestamp with microseconds"""
if isinstance(data, datetime.datetime):
@@ -70,6 +89,24 @@ def prepare_timestamp_micros(data, schema):
return data
+def prepare_local_timestamp_micros(
+ data: Union[datetime.datetime, int], schema: Dict
+) -> int:
+ """Converts datetime.datetime to int timestamp with microseconds
+
+ The local-timestamp-micros logical type represents a timestamp in a local
+ timezone, regardless of what specific time zone is considered local, with a
+ precision of one microsecond.
+ """
+ if isinstance(data, datetime.datetime):
+ delta = data.replace(tzinfo=datetime.timezone.utc) - epoch
+ return (
+ delta.days * 24 * 3600 + delta.seconds
+ ) * MCS_PER_SECOND + delta.microseconds
+ else:
+ return data
+
+
def prepare_date(data, schema):
"""Converts datetime.date to int timestamp"""
if isinstance(data, datetime.date):
@@ -212,7 +249,9 @@ def prepare_time_micros(data, schema):
LOGICAL_WRITERS = {
"long-timestamp-millis": prepare_timestamp_millis,
+ "long-local-timestamp-millis": prepare_local_timestamp_millis,
"long-timestamp-micros": prepare_timestamp_micros,
+ "long-local-timestamp-micros": prepare_local_timestamp_micros,
"int-date": prepare_date,
"bytes-decimal": prepare_bytes_decimal,
"fixed-decimal": prepare_fixed_decimal,

MODIFY
fastavro/_read.pyx
fastavro/_read.pyx
@@ -61,6 +61,7 @@ AVRO_TYPES = {
decimal_context = Context()
epoch = datetime(1970, 1, 1, tzinfo=timezone.utc)
+epoch_naive = datetime(1970, 1, 1)
ctypedef int int32
ctypedef unsigned int uint32
@@ -178,11 +179,21 @@ cpdef read_timestamp_millis(data, writer_schema=None, reader_schema=None):
return epoch + timedelta(microseconds=data * 1000)
+cpdef read_local_timestamp_millis(data, writer_schema=None, reader_schema=None):
+ # Cannot use datetime.fromtimestamp: https://bugs.python.org/issue36439
+ return epoch_naive + timedelta(microseconds=data * 1000)
+
+
cpdef read_timestamp_micros(data, writer_schema=None, reader_schema=None):
# Cannot use datetime.fromtimestamp: https://bugs.python.org/issue36439
return epoch + timedelta(microseconds=data)
+cpdef read_local_timestamp_micros(data, writer_schema=None, reader_schema=None):
+ # Cannot use datetime.fromtimestamp: https://bugs.python.org/issue36439
+ return epoch_naive + timedelta(microseconds=data)
+
+
cpdef read_date(data, writer_schema=None, reader_schema=None):
return date.fromordinal(data + DAYS_SHIFT)
@@ -716,7 +727,9 @@ cpdef skip_record(fo, writer_schema, named_schemas):
LOGICAL_READERS = {
"long-timestamp-millis": read_timestamp_millis,
+ "long-local-timestamp-millis": read_local_timestamp_millis,
"long-timestamp-micros": read_timestamp_micros,
+ "long-local-timestamp-micros": read_local_timestamp_micros,
"int-date": read_date,
"bytes-decimal": read_decimal,
"fixed-decimal": read_decimal,

MODIFY
fastavro/_read_py.py
fastavro/_read_py.py
@@ -61,6 +61,7 @@ AVRO_TYPES = {
decimal_context = Context()
epoch = datetime(1970, 1, 1, tzinfo=timezone.utc)
+epoch_naive = datetime(1970, 1, 1)
def match_types(writer_type, reader_type):
@@ -165,11 +166,25 @@ def read_timestamp_millis(data, writer_schema=None, reader_schema=None):
return epoch + timedelta(microseconds=data * 1000)
+def read_local_timestamp_millis(
+ data: int, writer_schema=None, reader_schema=None
+) -> datetime:
+ # Cannot use datetime.fromtimestamp: https://bugs.python.org/issue36439
+ return epoch_naive + timedelta(microseconds=data * 1000)
+
+
def read_timestamp_micros(data, writer_schema=None, reader_schema=None):
# Cannot use datetime.fromtimestamp: https://bugs.python.org/issue36439
return epoch + timedelta(microseconds=data)
+def read_local_timestamp_micros(
+ data: int, writer_schema=None, reader_schema=None
+) -> datetime:
+ # Cannot use datetime.fromtimestamp: https://bugs.python.org/issue36439
+ return epoch_naive + timedelta(microseconds=data)
+
+
def read_date(data, writer_schema=None, reader_schema=None):
return date.fromordinal(data + DAYS_SHIFT)
@@ -571,7 +586,9 @@ def skip_record(decoder, writer_schema, named_schemas):
LOGICAL_READERS = {
"long-timestamp-millis": read_timestamp_millis,
+ "long-local-timestamp-millis": read_local_timestamp_millis,
"long-timestamp-micros": read_timestamp_micros,
+ "long-local-timestamp-micros": read_local_timestamp_micros,
"int-date": read_date,
"bytes-decimal": read_decimal,
"fixed-decimal": read_decimal,

MODIFY
tests/test_logical_types.py
tests/test_logical_types.py
@@ -10,7 +10,7 @@ import datetime
import sys
import os
from dateutil.tz import tzlocal
-from datetime import timezone
+from datetime import timezone, timedelta
from .conftest import assert_naive_datetime_equal_to_tz_datetime
@@ -404,11 +404,7 @@ def test_default_scale_value():
def test_date_as_string():
- schema = {
- "name": "test_date_as_string",
- "type": "int",
- "logicalType": "date",
- }
+ schema = {"name": "test_date_as_string", "type": "int", "logicalType": "date"}
data1 = "2019-05-06"
binary = serialize(schema, data1)
@@ -442,3 +438,39 @@ def test_pandas_datetime():
)
}
assert serialize(schema, data1)
+
+
+def test_local_timestamp_millis():
+ schema = {"type": "long", "logicalType": "local-timestamp-millis"}
+
+ tz_naive = datetime.datetime(1970, 1, 1, 1)
+ binary = serialize(schema, tz_naive)
+ data2 = deserialize(schema, binary)
+ assert tz_naive == data2
+
+ tz_aware = datetime.datetime(1970, 1, 1, 1, tzinfo=timezone(timedelta(hours=5)))
+ binary = serialize(schema, tz_aware)
+ data2 = deserialize(schema, binary)
+ assert tz_naive == data2
+
+ binary = serialize(schema, 3600 * 1000)
+ data2 = deserialize(schema, binary)
+ assert tz_naive == data2
+
+
+def test_local_timestamp_micros():
+ schema = {"type": "long", "logicalType": "local-timestamp-micros"}
+
+ tz_naive = datetime.datetime(1970, 1, 1, 1)
+ binary = serialize(schema, tz_naive)
+ data2 = deserialize(schema, binary)
+ assert tz_naive == data2
+
+ tz_aware = datetime.datetime(1970, 1, 1, 1, tzinfo=timezone(timedelta(hours=5)))
+ binary = serialize(schema, tz_aware)
+ data2 = deserialize(schema, binary)
+ assert tz_naive == data2
+
+ binary = serialize(schema, 3600 * 1000 * 1000)
+ data2 = deserialize(schema, binary)
+ assert tz_naive == data2

add support for local-timestamp micros and millis