from ..adapt import Dumper, Loader
from ..proto import AdaptContext
+from ..errors import InterfaceError
from .oids import builtins
if self.connection:
ds = self.connection.pgconn.parameter_status(b"DateStyle")
if not ds or ds.startswith(b"ISO"):
- pass
+ pass # Default: YMD
elif ds.startswith(b"German"):
self.load = self.load_dmy # type: ignore
- elif ds.startswith(b"SQL"):
- self.load = self.load_mdy # type: ignore
- elif ds.startswith(b"Postgres"):
+ elif ds.startswith(b"SQL") or ds.startswith(b"Postgres"):
self.load = self.load_mdy # type: ignore
- def load(self, data: bytes) -> date:
- return date(int(data[:4]), int(data[5:7]), int(data[8:]))
+ def load_ymd(self, data: bytes) -> date:
+ try:
+ return date(int(data[:4]), int(data[5:7]), int(data[8:]))
+ except ValueError as e:
+ exc = e
+
+ return self._raise_error(data, exc)
+
+ load = load_ymd
def load_dmy(self, data: bytes) -> date:
- return date(int(data[6:]), int(data[3:5]), int(data[:2]))
+ try:
+ return date(int(data[6:]), int(data[3:5]), int(data[:2]))
+ except ValueError as e:
+ exc = e
+
+ return self._raise_error(data, exc)
def load_mdy(self, data: bytes) -> date:
- return date(int(data[6:]), int(data[:2]), int(data[3:5]))
+ try:
+ return date(int(data[6:]), int(data[:2]), int(data[3:5]))
+ except ValueError as e:
+ exc = e
+
+ return self._raise_error(data, exc)
+
+ def _raise_error(self, data: bytes, exc: ValueError) -> date:
+ # Most likely we received a BC date, which Python doesn't support
+ # Otherwise the unexpected value is displayed in the exception.
+ if data.endswith(b"BC"):
+ raise InterfaceError(
+ "BC date not supported by Python:"
+ f" {data.decode('utf8', 'replace')}")
+ else:
+ raise exc
import datetime as dt
import pytest
+import psycopg3
from psycopg3.adapt import Format
@pytest.mark.parametrize(
"val, expr",
[
+ (dt.date.min, "'0001-01-01'::date"),
(dt.date(1, 1, 1), "'0001-01-01'::date"),
(dt.date(1000, 1, 1), "'1000-01-01'::date"),
(dt.date(2000, 1, 1), "'2000-01-01'::date"),
(dt.date(2000, 12, 31), "'2000-12-31'::date"),
(dt.date(3000, 1, 1), "'3000-01-01'::date"),
+ (dt.date.max, "'9999-12-31'::date"),
],
)
def test_dump_date(conn, val, expr):
cur.execute(f"set datestyle = {datestyle_out}, YMD")
cur.execute("select '2000-01-02'::date")
assert cur.fetchone()[0] == dt.date(2000, 1, 2)
+
+
+@pytest.mark.parametrize("datestyle_out", ["ISO", "Postgres", "SQL", "German"])
+def test_load_date_bc(conn, datestyle_out):
+ cur = conn.cursor()
+ cur.execute(f"set datestyle = {datestyle_out}, YMD")
+ cur.execute("select '0001-01-01'::date - 1")
+ with pytest.raises(psycopg3.InterfaceError):
+ cur.fetchone()[0]