| 1964 | |
| 1965 | |
| 1966 | class _Timestamp(Timestamp): |
| 1967 | @classmethod |
| 1968 | def from_datetime(cls, dt: datetime) -> "_Timestamp": |
| 1969 | # manual epoch offset calulation to avoid rounding errors, |
| 1970 | # to support negative timestamps (before 1970) and skirt |
| 1971 | # around datetime bugs (apparently 0 isn't a year in [0, 9999]??) |
| 1972 | offset = dt - DATETIME_ZERO |
| 1973 | # below is the same as timedelta.total_seconds() but without dividing by 1e6 |
| 1974 | # so we end up with microseconds as integers instead of seconds as float |
| 1975 | offset_us = ( |
| 1976 | offset.days * 24 * 60 * 60 + offset.seconds |
| 1977 | ) * 10**6 + offset.microseconds |
| 1978 | seconds, us = divmod(offset_us, 10**6) |
| 1979 | return cls(seconds, us * 1000) |
| 1980 | |
| 1981 | def to_datetime(self) -> datetime: |
| 1982 | # datetime.fromtimestamp() expects a timestamp in seconds, not microseconds |
| 1983 | # if we pass it as a floating point number, we will run into rounding errors |
| 1984 | # see also #407 |
| 1985 | offset = timedelta(seconds=self.seconds, microseconds=self.nanos // 1000) |
| 1986 | return DATETIME_ZERO + offset |
| 1987 | |
| 1988 | @staticmethod |
| 1989 | def timestamp_to_json(dt: datetime) -> str: |
| 1990 | nanos = dt.microsecond * 1e3 |
| 1991 | if dt.tzinfo is not None: |
| 1992 | # change timezone aware datetime objects to utc |
| 1993 | dt = dt.astimezone(timezone.utc) |
| 1994 | copy = dt.replace(microsecond=0, tzinfo=None) |
| 1995 | result = copy.isoformat() |
| 1996 | if (nanos % 1e9) == 0: |
| 1997 | # If there are 0 fractional digits, the fractional |
| 1998 | # point '.' should be omitted when serializing. |
| 1999 | return f"{result}Z" |
| 2000 | if (nanos % 1e6) == 0: |
| 2001 | # Serialize 3 fractional digits. |
| 2002 | return f"{result}.{int(nanos // 1e6) :03d}Z" |
| 2003 | if (nanos % 1e3) == 0: |
| 2004 | # Serialize 6 fractional digits. |
| 2005 | return f"{result}.{int(nanos // 1e3) :06d}Z" |
| 2006 | # Serialize 9 fractional digits. |
| 2007 | return f"{result}.{nanos:09d}" |
| 2008 | |
| 2009 | |
| 2010 | def _get_wrapper(proto_type: str) -> Type: |