from .abc import Buffer, LoadFunc, AdaptContext, PyFormat, DumperKey
from .rows import Row, RowMaker
from ._compat import TypeAlias
-from .postgres import INVALID_OID
+from .postgres import INVALID_OID, TEXT_OID
from ._encodings import pgconn_encoding
if TYPE_CHECKING:
def as_literal(self, obj: Any) -> Buffer:
dumper = self.get_dumper(obj, PyFormat.TEXT)
rv = dumper.quote(obj)
- # If the result is quoted, and the oid not unknown,
+ # If the result is quoted, and the oid not unknown or text,
# add an explicit type cast.
# Check the last char because the first one might be 'E'.
oid = dumper.oid
- if oid and rv and rv[-1] == b"'"[0]:
+ if oid and rv and rv[-1] == b"'"[0] and oid != TEXT_OID:
try:
type_sql = self._oid_types[oid]
except KeyError:
rv = dumper.quote(obj)
oid = dumper.oid
- # If the result is quoted and the oid not unknown,
+ # If the result is quoted and the oid not unknown or text,
# add an explicit type cast.
# Check the last char because the first one might be 'E'.
- if oid and rv and rv[-1] == 39:
+ if oid and oid != oids.TEXT_OID and rv and rv[-1] == 39:
if self._oid_types is None:
self._oid_types = {}
type_ptr = PyDict_GetItem(<object>self._oid_types, oid)
assert cur.fetchone()[0] == "hellohello"
cur = conn.execute("select %s", [["hi", "ha"]])
assert cur.fetchone()[0] == ["hihi", "haha"]
- assert sql.Literal("hello").as_string(conn) == "'qelloqello'::text"
+ assert sql.Literal("hello").as_string(conn) == "'qelloqello'"
def test_loader_protocol(conn):
== "'{\"2000-01-01 00:00:00\"}'::timestamp[]"
)
+ def test_text_literal(self, conn):
+ conn.adapters.register_dumper(str, StrDumper)
+ assert sql.Literal("foo").as_string(conn) == "'foo'"
+
@pytest.mark.parametrize("name", ["a-b", f"{eur}", "order", "foo bar"])
def test_invalid_name(self, conn, name):
conn.execute(