class _BaseTimeDumper(Dumper):
-
- # Can change to timetz type if the object dumped is naive
- _oid = builtins["time"].oid
-
def get_key(
self, obj: time, format: Pg3Format
) -> Union[type, Tuple[type]]:
else:
return (self.cls,)
- def upgrade(self, obj: time, format: Pg3Format) -> "Dumper":
+ def upgrade(self, obj: time, format: Pg3Format) -> Dumper:
raise NotImplementedError
-class TimeDumper(_BaseTimeDumper):
+class _BaseTimeTextDumper(_BaseTimeDumper):
format = Format.TEXT
def dump(self, obj: time) -> bytes:
return str(obj).encode("utf8")
- def upgrade(self, obj: time, format: Pg3Format) -> "Dumper":
+
+class TimeDumper(_BaseTimeTextDumper):
+
+ _oid = builtins["time"].oid
+
+ def upgrade(self, obj: time, format: Pg3Format) -> Dumper:
if not obj.tzinfo:
return self
else:
return TimeTzDumper(self.cls)
-class TimeTzDumper(TimeDumper):
+class TimeTzDumper(_BaseTimeTextDumper):
_oid = builtins["timetz"].oid
class TimeBinaryDumper(_BaseTimeDumper):
format = Format.BINARY
+ _oid = builtins["time"].oid
def dump(self, obj: time) -> bytes:
ms = obj.microsecond + 1_000_000 * (
)
return _pack_int8(ms)
- def upgrade(self, obj: time, format: Pg3Format) -> "Dumper":
+ def upgrade(self, obj: time, format: Pg3Format) -> Dumper:
if not obj.tzinfo:
return self
else:
return TimeTzBinaryDumper(self.cls)
-class TimeTzBinaryDumper(TimeBinaryDumper):
+class TimeTzBinaryDumper(_BaseTimeDumper):
+ format = Format.BINARY
_oid = builtins["timetz"].oid
def dump(self, obj: time) -> bytes:
class _BaseDateTimeDumper(Dumper):
-
- # Can change to timestamp type if the object dumped is naive
- _oid = builtins["timestamptz"].oid
-
def get_key(
self, obj: datetime, format: Pg3Format
) -> Union[type, Tuple[type]]:
else:
return (self.cls,)
- def upgrade(self, obj: datetime, format: Pg3Format) -> "Dumper":
+ def upgrade(self, obj: datetime, format: Pg3Format) -> Dumper:
raise NotImplementedError
-class DateTimeTzDumper(_BaseDateTimeDumper):
+class _BaseDateTimeTextDumper(_BaseDateTimeDumper):
format = Format.TEXT
# the YYYY-MM-DD is always understood correctly.
return str(obj).encode("utf8")
- def upgrade(self, obj: datetime, format: Pg3Format) -> "Dumper":
+
+class DateTimeTzDumper(_BaseDateTimeTextDumper):
+
+ _oid = builtins["timestamptz"].oid
+
+ def upgrade(self, obj: datetime, format: Pg3Format) -> Dumper:
if obj.tzinfo:
return self
else:
return DateTimeDumper(self.cls)
-class DateTimeDumper(DateTimeTzDumper):
+class DateTimeDumper(_BaseDateTimeTextDumper):
+
_oid = builtins["timestamp"].oid
class DateTimeTzBinaryDumper(_BaseDateTimeDumper):
format = Format.BINARY
+ _oid = builtins["timestamptz"].oid
def dump(self, obj: datetime) -> bytes:
delta = obj - _pg_datetimetz_epoch
)
return _pack_int8(micros)
- def upgrade(self, obj: datetime, format: Pg3Format) -> "Dumper":
+ def upgrade(self, obj: datetime, format: Pg3Format) -> Dumper:
if obj.tzinfo:
return self
else:
return DateTimeBinaryDumper(self.cls)
-class DateTimeBinaryDumper(DateTimeTzBinaryDumper):
+class DateTimeBinaryDumper(_BaseDateTimeDumper):
+
+ format = Format.BINARY
_oid = builtins["timestamp"].oid
def dump(self, obj: datetime) -> bytes:
delta = obj - _pg_datetime_epoch
- micros = (
- 1_000_000 * (86_400 * delta.days + delta.seconds)
- + delta.microseconds
+ micros = delta.microseconds + 1_000_000 * (
+ 86_400 * delta.days + delta.seconds
)
return _pack_int8(micros)