import warnings
import threading
from types import TracebackType
-from typing import Any, AsyncIterator, Callable, Generic, Iterator, List
+from typing import Any, AsyncIterator, Dict, Callable, Generic, Iterator, List
from typing import NamedTuple, Optional, Type, TypeVar, Union
from typing import overload, TYPE_CHECKING
from weakref import ref, ReferenceType
connect = generators.connect
execute = generators.execute
-_UTC = ZoneInfo("UTC")
+_timezones: Dict[Union[None, bytes], ZoneInfo] = {}
class Notify(NamedTuple):
def timezone(self) -> ZoneInfo:
"""The Python timezone info of the connection's timezone."""
tzname = self.pgconn.parameter_status(b"TimeZone")
- if tzname:
+ try:
+ return _timezones[tzname]
+ except KeyError:
+ sname = tzname.decode("utf8") if tzname else "UTC"
try:
- return ZoneInfo(tzname.decode("utf8"))
+ zi = ZoneInfo(sname)
except KeyError:
logger.warning(
- "unknown PostgreSQL timezone: %r will use UTC",
- tzname.decode("utf8"),
+ "unknown PostgreSQL timezone: %r will use UTC", sname
)
- return _UTC
- else:
- return _UTC
+ zi = ZoneInfo("UTC")
+
+ _timezones[tzname] = zi
+ return zi
@property
def info(self) -> ConnectionInfo:
_pg_datetimetz_epoch = datetime(2000, 1, 1, tzinfo=timezone.utc)
_py_date_min_days = date.min.toordinal()
-_UTC = ZoneInfo("UTC")
-
class DateDumper(Dumper):
def __init__(self, oid: int, context: Optional[AdaptContext] = None):
super().__init__(oid, context)
- self._timezone = self.connection.timezone if self.connection else _UTC
+ self._timezone = (
+ self.connection.timezone if self.connection else ZoneInfo("UTC")
+ )
def _format_from_context(self) -> str:
ds = self._get_datestyle()
def __init__(self, oid: int, context: Optional[AdaptContext] = None):
super().__init__(oid, context)
- self._timezone = self.connection.timezone if self.connection else _UTC
+ self._timezone = (
+ self.connection.timezone if self.connection else ZoneInfo("UTC")
+ )
def load(self, data: Buffer) -> datetime:
micros = _unpack_int8(data)[0]