1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45
|
from __future__ import annotations
from psycopg import pq
from psycopg.abc import AdaptContext, Buffer, Dumper, Loader, PyFormat
def f() -> None:
d: Dumper = MyStrDumper(str, None)
assert d.dump("abc") == b"abcabc"
assert d.quote("abc") == b"'abcabc'"
lo: Loader = MyTextLoader(0, None)
assert lo.load(b"abc") == "abcabc"
class MyStrDumper:
format = pq.Format.TEXT
oid = 25 # text
def __init__(self, cls: type, context: AdaptContext | None = None):
self._cls = cls
def dump(self, obj: str) -> bytes:
return (obj * 2).encode()
def quote(self, obj: str) -> bytes:
value = self.dump(obj)
esc = pq.Escaping()
return b"'%s'" % esc.escape_string(value.replace(b"h", b"q"))
def get_key(self, obj: str, format: PyFormat) -> type:
return self._cls
def upgrade(self, obj: str, format: PyFormat) -> MyStrDumper:
return self
class MyTextLoader:
format = pq.Format.TEXT
def __init__(self, oid: int, context: AdaptContext | None = None):
pass
def load(self, data: Buffer) -> str:
return (bytes(data) * 2).decode()
|