From f0cdcc26ac9e9656cced8bd0c7e3d81fc23d5470 Mon Sep 17 00:00:00 2001
From: Claus Holbech <holbech@arista.com>
Date: Wed, 20 May 2026 17:16:32 +0200
Subject: [PATCH] Preserve nanosecond precision for Timestamp

Add NanoDatetime to carry sub-microsecond precision through
Timestamp conversions and JSON parsing/formatting while keeping
generated field types as datetime.datetime.

Generate the NanoDatetime import only for known types that need it,
and add runtime/compiler coverage for the new behavior.
---
 README.md                                     |   2 +-
 betterproto2/docs/getting-started.md          |   6 +
 betterproto2/docs/index.md                    |   1 +
 .../src/betterproto2/nano_datetime.py         | 163 ++++++++++++++++++
 betterproto2/tests/test_nano_datetime.py      |  83 +++++++++
 betterproto2/tests/test_timestamp.py          |  47 +++++
 .../known_types/__init__.py                   |   8 +-
 .../known_types/timestamp.py                  |  47 +----
 .../lib/google/protobuf/__init__.py           |  47 +----
 .../betterproto2_compiler/plugin/models.py    |  18 +-
 .../templates/header.py.j2                    |   4 +-
 .../tests/test_known_type_imports.py          |  17 ++
 plan.md                                       | 159 +++++++++++++++++
 13 files changed, 515 insertions(+), 87 deletions(-)
 create mode 100644 betterproto2/src/betterproto2/nano_datetime.py
 create mode 100644 betterproto2/tests/test_nano_datetime.py
 create mode 100644 betterproto2_compiler/tests/test_known_type_imports.py
 create mode 100644 plan.md

diff --git a/README.md b/README.md
index ce9b3190..0ab82a09 100644
--- a/README.md
+++ b/README.md
@@ -14,7 +14,7 @@ This project aims to provide an improved experience when using Protobuf / gRPC i
   - Enums
   - Dataclasses
   - `async`/`await`
-  - Timezone-aware `datetime` and `timedelta` objects
+  - Timezone-aware `datetime` and `timedelta` objects, including nanosecond-precision `Timestamp` values
   - Relative imports
   - Mypy type checking
 - [Pydantic Models](https://docs.pydantic.dev/) generation
diff --git a/betterproto2/docs/getting-started.md b/betterproto2/docs/getting-started.md
index 68d07495..2cb3621f 100644
--- a/betterproto2/docs/getting-started.md
+++ b/betterproto2/docs/getting-started.md
@@ -225,6 +225,12 @@ Message objects include `betterproto.Message.to_json` and
 `betterproto.Message.to_dict`, `betterproto.Message.from_dict` for
 converting back and forth from JSON serializable dicts.
 
+`google.protobuf.Timestamp` fields use timezone-aware `datetime.datetime`
+values. When binary or JSON data contains sub-microsecond precision,
+betterproto2 preserves it by returning a `betterproto2.nano_datetime.NanoDatetime`,
+which is a `datetime.datetime` subclass. Timestamp JSON accepts and emits
+RFC 3339 strings with up to 9 fractional second digits.
+
 For compatibility the default is to convert field names to
 `betterproto.Casing.CAMEL`. You can control this behavior by passing a
 different casing value, e.g:
diff --git a/betterproto2/docs/index.md b/betterproto2/docs/index.md
index 11fbfd6a..fa9864e0 100644
--- a/betterproto2/docs/index.md
+++ b/betterproto2/docs/index.md
@@ -12,5 +12,6 @@ Python code, using modern language features.
 
 - Generated messages are both binary & JSON serializable
 - Messages use relevant python types, e.g. ``Enum``, ``datetime`` and ``timedelta`` objects
+- ``Timestamp`` values preserve nanosecond precision when binary or JSON data contains it
 - ``async``/``await`` support for gRPC Clients and Servers
 - Generates modern, readable, idiomatic python code
diff --git a/betterproto2/src/betterproto2/nano_datetime.py b/betterproto2/src/betterproto2/nano_datetime.py
new file mode 100644
index 00000000..c2bebab4
--- /dev/null
+++ b/betterproto2/src/betterproto2/nano_datetime.py
@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import datetime
+import re
+from typing import Any
+
+import dateutil.parser
+from typing_extensions import Self
+
+_UTC = datetime.timezone.utc
+_TIMESTAMP_ZERO = datetime.datetime(1970, 1, 1, tzinfo=_UTC)
+_NANOS_PER_MICROSECOND = 1000
+_MICROS_PER_SECOND = 10**6
+_TIMESTAMP_RE = re.compile(
+    r"^"
+    r"(?P<date>\d{4}-\d{2}-\d{2})"
+    r"T"
+    r"(?P<time>\d{2}:\d{2}:\d{2})"
+    r"(?:\.(?P<fraction>\d{1,9}))?"
+    r"(?P<tz>Z|[+-]\d{2}:\d{2})"
+    r"$"
+)
+
+
+class NanoDatetime(datetime.datetime):
+    """A datetime that carries Timestamp sub-microsecond precision."""
+
+    __slots__ = ("_nanosecond_remainder",)
+
+    def __new__(
+        cls,
+        *args: Any,
+        nanosecond_remainder: int = 0,
+        **kwargs: Any,
+    ) -> Self:
+        if not 0 <= nanosecond_remainder < _NANOS_PER_MICROSECOND:
+            raise ValueError("nanosecond_remainder must be in range 0..999")
+
+        instance = super().__new__(cls, *args, **kwargs)
+        object.__setattr__(instance, "_nanosecond_remainder", nanosecond_remainder)
+        return instance
+
+    @property
+    def nanosecond_remainder(self) -> int:
+        return getattr(self, "_nanosecond_remainder", 0)
+
+    @property
+    def total_nanoseconds(self) -> int:
+        return self.microsecond * _NANOS_PER_MICROSECOND + self.nanosecond_remainder
+
+    def replace(
+        self, *args: Any, nanosecond_remainder: int | None = None, **kwargs: Any
+    ) -> NanoDatetime:
+        if nanosecond_remainder is None:
+            nanosecond_remainder = self.nanosecond_remainder
+
+        replaced = super().replace(*args, **kwargs)
+        return _datetime_to_nano_datetime(replaced, nanosecond_remainder)
+
+    def __repr__(self) -> str:
+        base = super().__repr__()
+        return f"{base[:-1]}, nanosecond_remainder={self.nanosecond_remainder})"
+
+    def __eq__(self, other: Any) -> bool:
+        equal = super().__eq__(other)
+        if equal is NotImplemented or not isinstance(other, datetime.datetime):
+            return equal
+
+        return bool(equal) and self.nanosecond_remainder == _datetime_nanosecond_remainder(other)
+
+    def __ne__(self, other: Any) -> bool:
+        equal = self.__eq__(other)
+        if equal is NotImplemented:
+            return equal
+
+        return not equal
+
+    def __hash__(self) -> int:
+        if self.nanosecond_remainder == 0:
+            return super().__hash__()
+
+        return hash((super().__hash__(), self.nanosecond_remainder))
+
+    @staticmethod
+    def to_timestamp(dt: datetime.datetime) -> tuple[int, int]:
+        if not dt.tzinfo:
+            raise ValueError("datetime must be timezone aware")
+
+        nanos = _datetime_total_nanoseconds(dt)
+        dt = dt.astimezone(_UTC)
+
+        offset = dt - _TIMESTAMP_ZERO
+        offset_us = (offset.days * 24 * 60 * 60 + offset.seconds) * _MICROS_PER_SECOND + offset.microseconds
+        seconds, _ = divmod(offset_us, _MICROS_PER_SECOND)
+        return seconds, nanos
+
+    @staticmethod
+    def from_timestamp(seconds: int, nanos: int) -> NanoDatetime:
+        micros, nanosecond_remainder = divmod(nanos, _NANOS_PER_MICROSECOND)
+        offset = datetime.timedelta(seconds=seconds, microseconds=micros)
+        dt = _TIMESTAMP_ZERO + offset
+        return _datetime_to_nano_datetime(dt, nanosecond_remainder)
+
+    @staticmethod
+    def to_json(dt: datetime.datetime) -> str:
+        seconds, nanos = NanoDatetime.to_timestamp(dt)
+        dt = NanoDatetime.from_timestamp(seconds, nanos)
+        result = dt.replace(microsecond=0, nanosecond_remainder=0, tzinfo=None).isoformat()
+
+        if nanos == 0:
+            return f"{result}Z"
+        if nanos % 1_000_000 == 0:
+            return f"{result}.{nanos // 1_000_000:03d}Z"
+        if nanos % 1_000 == 0:
+            return f"{result}.{nanos // 1_000:06d}Z"
+
+        return f"{result}.{nanos:09d}Z"
+
+    @staticmethod
+    def from_rfc3339(value: str) -> datetime.datetime:
+        match = _TIMESTAMP_RE.match(value)
+        if match:
+            tz = "+00:00" if match.group("tz") == "Z" else match.group("tz")
+            dt = datetime.datetime.fromisoformat(f"{match.group('date')}T{match.group('time')}{tz}")
+
+            fraction = (match.group("fraction") or "").ljust(9, "0")
+            total_nanos = int(fraction) if fraction else 0
+            micros, nanosecond_remainder = divmod(total_nanos, _NANOS_PER_MICROSECOND)
+
+            dt = dt.replace(microsecond=micros).astimezone(_UTC)
+            return _datetime_to_nano_datetime(dt, nanosecond_remainder)
+
+        dt = dateutil.parser.isoparse(value)
+        return dt.astimezone(_UTC)
+
+
+def _datetime_nanosecond_remainder(dt: datetime.datetime) -> int:
+    if isinstance(dt, NanoDatetime):
+        return dt.nanosecond_remainder
+
+    return 0
+
+
+def _datetime_total_nanoseconds(dt: datetime.datetime) -> int:
+    if isinstance(dt, NanoDatetime):
+        return dt.total_nanoseconds
+
+    return dt.microsecond * _NANOS_PER_MICROSECOND
+
+
+def _datetime_to_nano_datetime(dt: datetime.datetime, nanosecond_remainder: int = 0) -> NanoDatetime:
+    return NanoDatetime(
+        dt.year,
+        dt.month,
+        dt.day,
+        dt.hour,
+        dt.minute,
+        dt.second,
+        dt.microsecond,
+        tzinfo=dt.tzinfo,
+        fold=dt.fold,
+        nanosecond_remainder=nanosecond_remainder,
+    )
diff --git a/betterproto2/tests/test_nano_datetime.py b/betterproto2/tests/test_nano_datetime.py
new file mode 100644
index 00000000..a1b23efd
--- /dev/null
+++ b/betterproto2/tests/test_nano_datetime.py
@@ -0,0 +1,83 @@
+from datetime import datetime, timezone
+
+import pytest
+
+from betterproto2.nano_datetime import NanoDatetime
+
+
+@pytest.mark.parametrize("nanosecond_remainder", [-1, 1000])
+def test_nanosecond_remainder_must_be_sub_microsecond(nanosecond_remainder: int) -> None:
+    with pytest.raises(ValueError, match="nanosecond_remainder"):
+        NanoDatetime(2024, 1, 2, tzinfo=timezone.utc, nanosecond_remainder=nanosecond_remainder)
+
+
+def test_replace_preserves_nanosecond_remainder_by_default() -> None:
+    dt = NanoDatetime(2024, 1, 2, 3, 4, 5, 123456, tzinfo=timezone.utc, nanosecond_remainder=789)
+
+    replaced = dt.replace(second=6)
+
+    assert replaced == NanoDatetime(
+        2024,
+        1,
+        2,
+        3,
+        4,
+        6,
+        123456,
+        tzinfo=timezone.utc,
+        nanosecond_remainder=789,
+    )
+
+
+def test_replace_can_override_nanosecond_remainder() -> None:
+    dt = NanoDatetime(2024, 1, 2, 3, 4, 5, 123456, tzinfo=timezone.utc, nanosecond_remainder=789)
+
+    replaced = dt.replace(nanosecond_remainder=123)
+
+    assert replaced.nanosecond_remainder == 123
+    assert replaced.total_nanoseconds == 123456123
+
+
+def test_equality_and_hash_include_nanosecond_remainder() -> None:
+    base = datetime(2024, 1, 2, 3, 4, 5, 123456, tzinfo=timezone.utc)
+    zero_remainder = NanoDatetime(2024, 1, 2, 3, 4, 5, 123456, tzinfo=timezone.utc)
+    one_remainder = NanoDatetime(2024, 1, 2, 3, 4, 5, 123456, tzinfo=timezone.utc, nanosecond_remainder=1)
+    two_remainder = NanoDatetime(2024, 1, 2, 3, 4, 5, 123456, tzinfo=timezone.utc, nanosecond_remainder=2)
+
+    assert zero_remainder == base
+    assert hash(zero_remainder) == hash(base)
+    assert one_remainder != base
+    assert one_remainder != two_remainder
+    assert len({base, zero_remainder, one_remainder, two_remainder}) == 3
+
+
+@pytest.mark.parametrize(
+    ("dt", "expected"),
+    [
+        (NanoDatetime(1970, 1, 1, tzinfo=timezone.utc), "1970-01-01T00:00:00Z"),
+        (NanoDatetime(1970, 1, 1, 0, 0, 0, 123000, tzinfo=timezone.utc), "1970-01-01T00:00:00.123Z"),
+        (NanoDatetime(1970, 1, 1, 0, 0, 0, 123456, tzinfo=timezone.utc), "1970-01-01T00:00:00.123456Z"),
+        (
+            NanoDatetime(1970, 1, 1, 0, 0, 0, 123456, tzinfo=timezone.utc, nanosecond_remainder=789),
+            "1970-01-01T00:00:00.123456789Z",
+        ),
+    ],
+)
+def test_to_json_uses_required_fraction_widths(dt: NanoDatetime, expected: str) -> None:
+    assert NanoDatetime.to_json(dt) == expected
+
+
+@pytest.mark.parametrize(
+    ("value", "total_nanoseconds"),
+    [
+        ("1970-01-01T00:00:00.1Z", 100000000),
+        ("1970-01-01T00:00:00.1234Z", 123400000),
+        ("1970-01-01T00:00:00.12345678Z", 123456780),
+        ("1970-01-01T00:00:00.123456789Z", 123456789),
+    ],
+)
+def test_from_rfc3339_accepts_subsecond_fraction_widths(value: str, total_nanoseconds: int) -> None:
+    dt = NanoDatetime.from_rfc3339(value)
+
+    assert isinstance(dt, NanoDatetime)
+    assert dt.total_nanoseconds == total_nanoseconds
diff --git a/betterproto2/tests/test_timestamp.py b/betterproto2/tests/test_timestamp.py
index 8187939b..f4aab5d2 100644
--- a/betterproto2/tests/test_timestamp.py
+++ b/betterproto2/tests/test_timestamp.py
@@ -5,6 +5,7 @@
 
 import pytest
 
+from betterproto2.nano_datetime import NanoDatetime
 from tests.outputs.google.google.protobuf import Timestamp
 
 
@@ -34,3 +35,49 @@ def test_invalid_datetime():
     """
     with pytest.raises(ValueError):
         Timestamp.from_datetime(datetime.now())
+
+
+def test_timestamp_to_datetime_preserves_nanoseconds():
+    ts = Timestamp(seconds=1, nanos=123456789)
+
+    dt = ts.to_datetime()
+
+    assert isinstance(dt, NanoDatetime)
+    assert dt.microsecond == 123456
+    assert dt.nanosecond_remainder == 789
+    assert dt.total_nanoseconds == 123456789
+    assert Timestamp.from_datetime(dt) == ts
+
+
+def test_timestamp_to_datetime_preserves_negative_nanoseconds():
+    ts = Timestamp(seconds=-1, nanos=999999999)
+
+    dt = ts.to_datetime()
+
+    assert isinstance(dt, NanoDatetime)
+    assert dt == NanoDatetime(
+        1969,
+        12,
+        31,
+        23,
+        59,
+        59,
+        999999,
+        tzinfo=timezone.utc,
+        nanosecond_remainder=999,
+    )
+    assert Timestamp.from_datetime(dt) == ts
+
+
+def test_timestamp_dict_preserves_nanoseconds():
+    ts = Timestamp.from_dict("1970-01-01T00:00:01.123456789Z")
+
+    assert ts == Timestamp(seconds=1, nanos=123456789)
+    assert ts.to_dict() == "1970-01-01T00:00:01.123456789Z"
+
+
+def test_timestamp_dict_preserves_nanoseconds_with_offset():
+    ts = Timestamp.from_dict("1970-01-01T01:00:01.123456789+01:00")
+
+    assert ts == Timestamp(seconds=1, nanos=123456789)
+    assert ts.to_dict() == "1970-01-01T00:00:01.123456789Z"
diff --git a/betterproto2_compiler/src/betterproto2_compiler/known_types/__init__.py b/betterproto2_compiler/src/betterproto2_compiler/known_types/__init__.py
index 5ca6e9e4..3966da21 100644
--- a/betterproto2_compiler/src/betterproto2_compiler/known_types/__init__.py
+++ b/betterproto2_compiler/src/betterproto2_compiler/known_types/__init__.py
@@ -17,8 +17,7 @@
 from .timestamp import Timestamp
 
 # For each (package, message name), lists the methods that should be added to the message definition.
-# The source code of the method is read from the `known_types` folder. If imports are needed, they can be directly added
-# to the template file: they will automatically be removed if not necessary.
+# The source code of the method is read from the `known_types` folder.
 KNOWN_METHODS: dict[tuple[str, str], list[Callable]] = {
     ("google.protobuf", "Any"): [Any.pack, Any.unpack, Any.to_dict, Any.from_dict],
     ("google.protobuf", "Timestamp"): [
@@ -107,6 +106,11 @@
     ],
 }
 
+# For each (package, message name), lists imports required by known-type methods.
+KNOWN_IMPORTS: dict[tuple[str, str], tuple[str, ...]] = {
+    ("google.protobuf", "Timestamp"): ("from betterproto2.nano_datetime import NanoDatetime",),
+}
+
 # A wrapped type is the type of a message that is automatically replaced by a known Python type.
 WRAPPED_TYPES: dict[tuple[str, str], str] = {
     ("google.protobuf", "BoolValue"): "bool",
diff --git a/betterproto2_compiler/src/betterproto2_compiler/known_types/timestamp.py b/betterproto2_compiler/src/betterproto2_compiler/known_types/timestamp.py
index bd8f6e79..30ab62c4 100644
--- a/betterproto2_compiler/src/betterproto2_compiler/known_types/timestamp.py
+++ b/betterproto2_compiler/src/betterproto2_compiler/known_types/timestamp.py
@@ -2,7 +2,7 @@
 import typing
 
 import betterproto2
-import dateutil.parser
+from betterproto2.nano_datetime import NanoDatetime
 from typing_extensions import Self
 
 from betterproto2_compiler.lib.google.protobuf import Timestamp as VanillaTimestamp
@@ -11,56 +11,21 @@
 class Timestamp(VanillaTimestamp):
     @classmethod
     def from_datetime(cls, dt: datetime.datetime) -> Self:
-        if not dt.tzinfo:
-            raise ValueError("datetime must be timezone aware")
-
-        dt = dt.astimezone(datetime.timezone.utc)
-
-        # manual epoch offset calulation to avoid rounding errors,
-        # to support negative timestamps (before 1970) and skirt
-        # around datetime bugs (apparently 0 isn't a year in [0, 9999]??)
-        offset = dt - datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
-        # below is the same as timedelta.total_seconds() but without dividing by 1e6
-        # so we end up with microseconds as integers instead of seconds as float
-        offset_us = (offset.days * 24 * 60 * 60 + offset.seconds) * 10**6 + offset.microseconds
-        seconds, us = divmod(offset_us, 10**6)
-        return cls(seconds, us * 1000)
+        seconds, nanos = NanoDatetime.to_timestamp(dt)
+        return cls(seconds, nanos)
 
     def to_datetime(self) -> datetime.datetime:
-        # datetime.fromtimestamp() expects a timestamp in seconds, not microseconds
-        # if we pass it as a floating point number, we will run into rounding errors
-        # see also #407
-        offset = datetime.timedelta(seconds=self.seconds, microseconds=self.nanos // 1000)
-        return datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc) + offset
+        return NanoDatetime.from_timestamp(self.seconds, self.nanos)
 
     @staticmethod
     def timestamp_to_json(dt: datetime.datetime) -> str:
-        nanos = dt.microsecond * 1e3
-        if dt.tzinfo is not None:
-            # change timezone aware datetime objects to utc
-            dt = dt.astimezone(datetime.timezone.utc)
-        copy = dt.replace(microsecond=0, tzinfo=None)
-        result = copy.isoformat()
-        if (nanos % 1e9) == 0:
-            # If there are 0 fractional digits, the fractional
-            # point '.' should be omitted when serializing.
-            return f"{result}Z"
-        if (nanos % 1e6) == 0:
-            # Serialize 3 fractional digits.
-            return f"{result}.{int(nanos // 1e6):03d}Z"
-        if (nanos % 1e3) == 0:
-            # Serialize 6 fractional digits.
-            return f"{result}.{int(nanos // 1e3):06d}Z"
-        # Serialize 9 fractional digits.
-        return f"{result}.{nanos:09d}"
+        return NanoDatetime.to_json(dt)
 
     # TODO typing
     @classmethod
     def from_dict(cls, value, *, ignore_unknown_fields: bool = False) -> Self:
         if isinstance(value, str):
-            dt = dateutil.parser.isoparse(value)
-            dt = dt.astimezone(datetime.timezone.utc)
-            return cls.from_datetime(dt)
+            return cls.from_datetime(NanoDatetime.from_rfc3339(value))
 
         return super().from_dict(value, ignore_unknown_fields=ignore_unknown_fields)
 
diff --git a/betterproto2_compiler/src/betterproto2_compiler/lib/google/protobuf/__init__.py b/betterproto2_compiler/src/betterproto2_compiler/lib/google/protobuf/__init__.py
index 9652c5fc..12c0a311 100644
--- a/betterproto2_compiler/src/betterproto2_compiler/lib/google/protobuf/__init__.py
+++ b/betterproto2_compiler/src/betterproto2_compiler/lib/google/protobuf/__init__.py
@@ -93,7 +93,7 @@
 from dataclasses import dataclass
 
 import betterproto2
-import dateutil.parser
+from betterproto2.nano_datetime import NanoDatetime
 from typing_extensions import Self
 
 from ...message_pool import default_message_pool
@@ -3580,55 +3580,20 @@ class Timestamp(betterproto2.Message):
 
     @classmethod
     def from_datetime(cls, dt: datetime.datetime) -> Self:
-        if not dt.tzinfo:
-            raise ValueError("datetime must be timezone aware")
-
-        dt = dt.astimezone(datetime.timezone.utc)
-
-        # manual epoch offset calulation to avoid rounding errors,
-        # to support negative timestamps (before 1970) and skirt
-        # around datetime bugs (apparently 0 isn't a year in [0, 9999]??)
-        offset = dt - datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
-        # below is the same as timedelta.total_seconds() but without dividing by 1e6
-        # so we end up with microseconds as integers instead of seconds as float
-        offset_us = (offset.days * 24 * 60 * 60 + offset.seconds) * 10**6 + offset.microseconds
-        seconds, us = divmod(offset_us, 10**6)
-        return cls(seconds, us * 1000)
+        seconds, nanos = NanoDatetime.to_timestamp(dt)
+        return cls(seconds, nanos)
 
     def to_datetime(self) -> datetime.datetime:
-        # datetime.fromtimestamp() expects a timestamp in seconds, not microseconds
-        # if we pass it as a floating point number, we will run into rounding errors
-        # see also #407
-        offset = datetime.timedelta(seconds=self.seconds, microseconds=self.nanos // 1000)
-        return datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc) + offset
+        return NanoDatetime.from_timestamp(self.seconds, self.nanos)
 
     @staticmethod
     def timestamp_to_json(dt: datetime.datetime) -> str:
-        nanos = dt.microsecond * 1e3
-        if dt.tzinfo is not None:
-            # change timezone aware datetime objects to utc
-            dt = dt.astimezone(datetime.timezone.utc)
-        copy = dt.replace(microsecond=0, tzinfo=None)
-        result = copy.isoformat()
-        if (nanos % 1e9) == 0:
-            # If there are 0 fractional digits, the fractional
-            # point '.' should be omitted when serializing.
-            return f"{result}Z"
-        if (nanos % 1e6) == 0:
-            # Serialize 3 fractional digits.
-            return f"{result}.{int(nanos // 1e6):03d}Z"
-        if (nanos % 1e3) == 0:
-            # Serialize 6 fractional digits.
-            return f"{result}.{int(nanos // 1e3):06d}Z"
-        # Serialize 9 fractional digits.
-        return f"{result}.{nanos:09d}"
+        return NanoDatetime.to_json(dt)
 
     @classmethod
     def from_dict(cls, value, *, ignore_unknown_fields: bool = False) -> Self:
         if isinstance(value, str):
-            dt = dateutil.parser.isoparse(value)
-            dt = dt.astimezone(datetime.timezone.utc)
-            return cls.from_datetime(dt)
+            return cls.from_datetime(NanoDatetime.from_rfc3339(value))
 
         return super().from_dict(value, ignore_unknown_fields=ignore_unknown_fields)
 
diff --git a/betterproto2_compiler/src/betterproto2_compiler/plugin/models.py b/betterproto2_compiler/src/betterproto2_compiler/plugin/models.py
index 085e43b0..0f27e5e7 100644
--- a/betterproto2_compiler/src/betterproto2_compiler/plugin/models.py
+++ b/betterproto2_compiler/src/betterproto2_compiler/plugin/models.py
@@ -38,7 +38,7 @@
     pythonize_field_name,
     pythonize_method_name,
 )
-from betterproto2_compiler.known_types import KNOWN_METHODS, WRAPPED_TYPES
+from betterproto2_compiler.known_types import KNOWN_IMPORTS, KNOWN_METHODS, WRAPPED_TYPES
 from betterproto2_compiler.lib.google.protobuf import (
     DescriptorProto,
     EnumDescriptorProto,
@@ -241,6 +241,15 @@ def descriptors(self):
 
         return "\n".join(descriptors)
 
+    @property
+    def custom_imports(self) -> list[str]:
+        imports: set[str] = set()
+
+        for message in self.messages.values():
+            imports.update(message.custom_imports)
+
+        return sorted(imports)
+
 
 @dataclass(kw_only=True)
 class MessageCompiler(ProtoContentBase):
@@ -292,6 +301,13 @@ def custom_methods(self) -> list[str]:
 
         return methods_source
 
+    @property
+    def custom_imports(self) -> tuple[str, ...]:
+        """
+        Return imports required by custom known-type methods.
+        """
+        return KNOWN_IMPORTS.get((self.source_file.package, self.py_name), ())
+
     @property
     def descriptor_name(self) -> str:
         """Google protobuf library descriptor name.
diff --git a/betterproto2_compiler/src/betterproto2_compiler/templates/header.py.j2 b/betterproto2_compiler/src/betterproto2_compiler/templates/header.py.j2
index 750e1f66..752f437d 100644
--- a/betterproto2_compiler/src/betterproto2_compiler/templates/header.py.j2
+++ b/betterproto2_compiler/src/betterproto2_compiler/templates/header.py.j2
@@ -14,7 +14,6 @@ __all__ = (
 import re
 import builtins
 import datetime
-import dateutil.parser
 import warnings
 from collections.abc import AsyncIterable, AsyncIterator, Iterable, Iterator
 import typing
@@ -34,6 +33,9 @@ from betterproto2 import grpclib as betterproto2_grpclib
 import grpc
 import grpclib
 from google.protobuf.descriptor import Descriptor, EnumDescriptor
+{% for custom_import in output_file.custom_imports %}
+{{ custom_import }}
+{% endfor %}
 
 {# Import the message pool of the generated code. #}
 {% if output_file.package %}
diff --git a/betterproto2_compiler/tests/test_known_type_imports.py b/betterproto2_compiler/tests/test_known_type_imports.py
new file mode 100644
index 00000000..1ad95349
--- /dev/null
+++ b/betterproto2_compiler/tests/test_known_type_imports.py
@@ -0,0 +1,17 @@
+from pathlib import Path
+
+OUTPUTS = Path(__file__).parent / "outputs"
+
+
+def test_timestamp_output_imports_nano_datetime() -> None:
+    generated = (OUTPUTS / "google" / "google" / "protobuf" / "__init__.py").read_text()
+
+    assert "from betterproto2.nano_datetime import NanoDatetime" in generated
+    assert "NanoDatetime.from_timestamp" in generated
+    assert "dateutil.parser" not in generated
+
+
+def test_non_timestamp_output_does_not_import_nano_datetime() -> None:
+    generated = (OUTPUTS / "bool" / "bool" / "__init__.py").read_text()
+
+    assert "betterproto2.nano_datetime" not in generated
diff --git a/plan.md b/plan.md
new file mode 100644
index 00000000..fa9204a2
--- /dev/null
+++ b/plan.md
@@ -0,0 +1,159 @@
+Porting your nanosecond-precision implementation from `aristaproto` into the modern `betterproto2` architecture provides a great opportunity to make the code cleaner, more efficient, and more resilient.
+
+Below is an analysis of how to optimize the design of the `datetime` subclass and exactly where it needs to hook into the `betterproto2` core runtime and compiler.
+
+---
+
+## 1. Tightening the Class Shape
+
+In `aristaproto`, a `datetime` subclass works beautifully because it maintains perfect backward compatibility with standard `isinstance(dt, datetime)` checks. However, when parsing millions of telemetry records, a naive subclass can introduce serious memory and performance overhead.
+
+To tighten the shape for `betterproto2`, three critical optimizations are recommended:
+
+### Use `__slots__` to Prevent Memory Bloat
+
+By default, Python subclasses of built-in types create a local instance dictionary (`__dict__`). If you instantiate millions of timestamps from a gRPC stream, this destroys memory efficiency. Using `__slots__` drops the memory footprint back to native-C levels.
+
+### Immutable Remainder Handling
+
+Instead of making the nanosecond remainder mutable or recalculating it constantly, capture it exactly once during `__new__`.
+
+### Tightened Implementation Concept
+
+```python
+from datetime import datetime, timezone
+
+class nanodatetime(datetime):
+    # __slots__ eliminates __dict__ and minimizes memory footprint per instance
+    __slots__ = ("_nanos_remainder",)
+
+    def __new__(cls, *args, **kwargs):
+        # Pop the nanosecond remainder (0-999) before passing to native datetime
+        nanos_remainder = kwargs.pop("nanos_remainder", 0)
+        
+        # Instantiate the standard microsecond-bounded datetime object
+        instance = super().__new__(cls, *args, **kwargs)
+        
+        # Attach the immutable high-precision remainder
+        object.__setattr__(instance, "_nanos_remainder", nanos_remainder)
+        return instance
+
+    @property
+    def nanosecond_remainder(self) -> int:
+        """Returns the sub-microsecond remainder (0 to 999 nanoseconds)."""
+        return self._nanos_remainder
+
+    @property
+    def total_nanoseconds(self) -> int:
+        """Returns the exact trailing nanoseconds within the current second (0 to 999,999,999)."""
+        return (self.microsecond * 1000) + self._nanos_remainder
+
+    def __repr__(self) -> str:
+        # Extend repr to show full nanosecond precision transparently
+        base = super().__repr__()
+        return f"{base[:-1]}, nanos_remainder={self._nanos_remainder})"
+
+```
+
+---
+
+## 2. Identifying Integration Points in `betterproto2`
+
+In `betterproto2`, Well-Known Types (WKT) like `google.protobuf.Timestamp` are handled differently than normal message fields. They have dedicated hooks inside the binary and dictionary codecs.
+
+You will need to touch exactly **three integration spots** in the upstream repository:
+
+### Integration Point A: Binary Deserialization (`betterproto2/runtime/pack.py` or `runtime/codec.py`)
+
+Look for the parsing engine that decodes incoming wire fields. When `betterproto2` finishes pulling the `seconds` (int64) and `nanos` (int32) out of the wire for a `Timestamp` message, it immediately turns them into a standard datetime.
+
+This is your first injection point:
+
+```python
+# CURRENT BETTERPROTO2 FALLBACK CONCEPT:
+# dt = datetime.fromtimestamp(seconds, tz=timezone.utc).replace(microsecond=nanos // 1000)
+
+# TIGHTENED UPSTREAM PORT:
+micros, nanos_remainder = divmod(nanos, 1000)
+dt = nanodatetime.fromtimestamp(seconds, tz=timezone.utc).replace(
+    microsecond=micros, 
+    nanos_remainder=nanos_remainder
+)
+
+```
+
+### Integration Point B: Binary Serialization
+
+When a user passes a datetime object back into a message to write it to the wire, `betterproto2` strips out the data. You need to inspect the object type to capture the remainder safely.
+
+```python
+def datetime_to_timestamp_fields(dt: datetime) -> tuple[int, int]:
+    seconds = int(dt.timestamp())
+    
+    # Check if it's our custom high-precision subclass
+    if isinstance(dt, nanodatetime):
+        nanos = dt.total_nanoseconds
+    else:
+        nanos = dt.microsecond * 1000
+        
+    return seconds, nanos
+
+```
+
+### Integration Point C: JSON/Dictionary Codecs (`betterproto2/runtime/mixins.py` or `to_dict` logic)
+
+According to the official Google specification, JSON representations of RFC 3339 timestamps **must** support up to 9 digits of sub-second precision (e.g., `2026-05-20T12:00:00.123456789Z`).
+
+* Update the `to_dict()` encoder method for timestamps to grab `dt.total_nanoseconds` if available, formatting the string dynamically instead of relying purely on `.isoformat()` (which caps out at 6 digits for microseconds).
+* Update `from_dict()` to regex-parse incoming RFC strings to split strings with >6 decimal places into microseconds and your custom `nanos_remainder`.
+
+---
+
+## 3. The Compiler Consideration
+
+In `betterproto2_compiler`, fields mapping to `google.protobuf.Timestamp` generate type hints matching `datetime`.
+
+> **Tip:** Keep the generated type hints exactly as `datetime`. Because your `nanodatetime` is a literal subclass, IDE type-checkers (like Mypy or Pyright) will accept it transparently without creating extra import errors or requiring users to add `from betterproto2 import nanodatetime` all over their business logic.
+
+---
+That makes complete sense. In network and device telemetry (especially the kind of streaming states Arista switches handle), absolute **`Timestamp`** data is the core driver. `Duration` is rarely used at sub-microsecond precision in those pipelines, so skipping it keeps your upstream PR tightly scoped, focused, and much easier for the maintainers to review and approve.
+
+By focusing purely on `Timestamp`, you can cleanly isolate the integration points within the **`betterproto/python-betterproto2`** repository.
+
+---
+
+## Where to Look in `betterproto2`
+
+The `betterproto2` repository has a clean split: `betterproto2_compiler/` handles the code-generation plugin, while `betterproto2/` handles the runtime. You will only need to modify files inside the **`betterproto2/`** package runtime.
+
+Here are the exact file paths and code mechanisms you will want to target for your PR:
+
+### 1. The Core WKT Mappings
+
+* **File to target:** `betterproto2/lib/google/protobuf/__init__.py` (or the equivalent generated WKT stub folder inside the runtime).
+* **The Logic:** In `betterproto`, `google.protobuf.Timestamp` is treated as a specialized message class that inherits from a base `Message` class but intercepts binary serialization/deserialization. You will want to define your tight `nanodatetime` subclass at the top of this file (or in a core utility module) and swap out the default `datetime.fromtimestamp(...)` instantiation for your custom type.
+
+### 2. JSON / Dictionary Serialization
+
+* **File to target:** `betterproto2/runtime/mixins.py` (specifically look for the `to_dict` and `from_dict` implementations).
+* **The Logic:** Protobuf's official JSON specification requires `Timestamp` to be encoded as an RFC 3339 string.
+* In `to_dict`, locate the conditional block checking if a message is a `google.protobuf.Timestamp`. Replace the basic `.isoformat()` serialization with logic that explicitly formats the full 9 digits using your `total_nanoseconds` property if the instance is a `nanodatetime`.
+* In `from_dict`, update the string parsing to capture any sub-microsecond digits (places 7, 8, and 9 after the decimal point) instead of discarding them.
+
+
+
+### 3. Binary Packing Logic
+
+* **File to target:** `betterproto2/runtime/codec.py` or `betterproto2/runtime/pack.py`.
+* **The Logic:** This handles low-level serialization to bytes. Ensure that when a field value is converted to a `Timestamp` on its way to the wire, the engine extracts the sub-microsecond remainder using `isinstance(val, nanodatetime)` instead of multiplying the default `.microsecond * 1000`.
+
+---
+
+## Strategy for a Smooth Upstream Merge
+
+Because you are writing this directly for `betterproto2`, you can make your implementation look exceptionally native. Here are a couple of stylistic tips to guarantee the maintainers accept the PR without friction:
+
+* **Keep Type Hints Generic:** Do not change the generated code's type hints from `datetime` to `nanodatetime`. Because your class is a direct subclass of `datetime`, standard typing engines accept it flawlessly. This keeps the compiler output pristine and leaves zero footprint for users who don't care about nanoseconds.
+* **Add Explicit Tests:** Add a test case in the test suite that specifically serializes a message containing exactly `123456789` nanoseconds, sends it over the wire (or dumps it to a dict), and asserts that it decodes back into the exact same value without rounding or dropping the final three digits.
+
+This is going to be a stellar upgrade for the library. Let me know if you want to draft the exact `to_dict` JSON parsing block together once you pull the repo!
\ No newline at end of file