self.adapters: AdaptersMap = {}
self.casters: TypecastersMap = {}
# name of the postgres encoding (in bytes)
- self.pgenc = b""
+ self._pgenc = b""
def cursor(
self, name: Optional[str] = None, binary: bool = False
def codec(self) -> codecs.CodecInfo:
# TODO: utf8 fastpath?
pgenc = self.pgconn.parameter_status(b"client_encoding")
- if self.pgenc != pgenc:
- # for unknown encodings and SQL_ASCII be strict and use ascii
- pyenc = pq.py_codecs.get(pgenc.decode("ascii")) or "ascii"
+ if self._pgenc != pgenc:
+ try:
+ pyenc = pq.py_codecs[pgenc.decode("ascii")]
+ except KeyError:
+ raise e.NotSupportedError(
+ f"encoding {pgenc.decode('ascii')} not available in Python"
+ )
self._codec = codecs.lookup(pyenc)
- self.pgenc = pgenc
+ self._pgenc = pgenc
return self._codec
def encode(self, s: str) -> bytes:
# "MULE_INTERNAL": not available in Python
"SHIFT_JIS_2004": "shift_jis_2004",
"SJIS": "shift_jis",
- "SQL_ASCII": None, # means no encoding, see PostgreSQL docs
+ # this actually means no encoding, see PostgreSQL docs
+ # it is special-cased by the text typecaster.
+ "SQL_ASCII": "ascii",
"UHC": "cp949",
"UTF8": "utf-8",
"WIN1250": "cp1250",
assert enc == newenc
+def test_set_encoding_unsupported(conn):
+ conn.encoding = "EUC_TW"
+ with pytest.raises(psycopg3.NotSupportedError):
+ conn.cursor().execute("select 1")
+
+
def test_set_encoding_bad(conn):
with pytest.raises(psycopg3.DatabaseError):
conn.encoding = "WAT"