(for instance, a Python `int` might be better dumped as a PostgreSQL
:sql:`integer`, :sql:`bigint`, :sql:`smallint` according to its value).
+- According to the placeholder used (``%s``, ``%b``, ``%t``), psycopg3 may
+ pick a binary or a text dumper. When using the ``%s`` "`~Format.AUTO`"
+ format, if the same type has both a text and a binary dumper registered, the
+ last one registered (using `Dumper.register()`) will be selected.
+
- For every OID returned by the query, the `!Transformer` will instantiate a
`!Loader`. All the values with the same OID will be converted by the same
loader.
You should call this method on the `Dumper` subclass you create,
passing the Python type you want to dump as *cls*.
+ If two dumpers of different `format` are registered for the same type,
+ the last one registered will be chosen by default when the query
+ doesn't specify a format (i.e. when the value is used with a ``%s``
+ "`~Format.AUTO`" placeholder).
+
:param cls: The type to manage.
:type cls: `!type` or `!str`
:param context: Where the dumper should be used. If `!None` the dumper
`~psycopg3.pq.Format.BINARY`, available most of the times. Usually the binary
format is more efficient to use.
-`~psycopg3` can support both the formats of each data type, and normally will
-use the binary type if available, falling back on the text type. The selection
-is automatic and will be performed whenever a query uses a ``%s`` placeholder
-to specify the value.
+`~psycopg3` can support both the formats of each data type. Whenever a value
+is passed to a query using the normal ``%s`` placeholder, the best format
+available is chosen (often, but not always, the binary format is picked as the
+best choice).
If you have a reason to select explicitly the binary format or the text format
for a value you can use respectively a ``%b`` placeholder or a ``%t``
is cheap: a copy is made only on customisation.
"""
- _dumpers: List[Dict[Union[type, str], Type["Dumper"]]]
+ _dumpers: Dict[Format, Dict[Union[type, str], Type["Dumper"]]]
_loaders: List[Dict[int, Type["Loader"]]]
types: TypesRegistry
types: Optional[TypesRegistry] = None,
):
if template:
- self._dumpers = template._dumpers[:]
- self._own_dumpers = [False, False]
+ self._dumpers = template._dumpers.copy()
+ self._own_dumpers = dict.fromkeys(Format, False)
self._loaders = template._loaders[:]
self._own_loaders = [False, False]
self.types = TypesRegistry(template.types)
else:
- self._dumpers = [{}, {}]
- self._own_dumpers = [True, True]
+ self._dumpers = {fmt: {} for fmt in Format}
+ self._own_dumpers = dict.fromkeys(Format, True)
self._loaders = [{}, {}]
self._own_loaders = [True, True]
self.types = types or TypesRegistry()
)
dumper = self._get_optimised(dumper)
- fmt = dumper.format
- if not self._own_dumpers[fmt]:
- self._dumpers[fmt] = self._dumpers[fmt].copy()
- self._own_dumpers[fmt] = True
+ # Register the dumper both as its format and as default
+ for fmt in (Format.from_pq(dumper.format), Format.AUTO):
+ if not self._own_dumpers[fmt]:
+ self._dumpers[fmt] = self._dumpers[fmt].copy()
+ self._own_dumpers[fmt] = True
- self._dumpers[fmt][cls] = dumper
+ self._dumpers[fmt][cls] = dumper
def register_loader(
self, oid: Union[int, str], loader: Type[Loader]
Raise ProgrammingError if a class is not available.
"""
- if format == Format.AUTO:
- # When dumping a string with %s we may refer to any type actually,
- # but the user surely passed a text format
- if issubclass(cls, str):
- dmaps = [self._dumpers[pq.Format.TEXT]]
- else:
- dmaps = [
- self._dumpers[pq.Format.BINARY],
- self._dumpers[pq.Format.TEXT],
- ]
- elif format == Format.BINARY:
- dmaps = [self._dumpers[pq.Format.BINARY]]
- elif format == Format.TEXT:
- dmaps = [self._dumpers[pq.Format.TEXT]]
- else:
+ try:
+ dmap = self._dumpers[format]
+ except KeyError:
raise ValueError(f"bad dumper format: {format}")
# Look for the right class, including looking at superclasses
for scls in cls.__mro__:
- for dmap in dmaps:
- if scls in dmap:
- return dmap[scls]
+ if scls in dmap:
+ return dmap[scls]
# If the adapter is not found, look for its name as a string
fqn = scls.__module__ + "." + scls.__qualname__
- for dmap in dmaps:
- if fqn in dmap:
- # Replace the class name with the class itself
- d = dmap[scls] = dmap.pop(fqn)
- return d
+ if fqn in dmap:
+ # Replace the class name with the class itself
+ d = dmap[scls] = dmap.pop(fqn)
+ return d
raise e.ProgrammingError(
f"cannot adapt type {cls.__name__}"
def register_default_globals(ctx: AdaptContext) -> None:
- StringDumper.register(str, ctx)
+ # NOTE: the order the dumpers are registered is relevant.
+ # The last one registered becomes the default for each type.
+ # Normally, binary is the default dumper, except for text (which plays
+ # the role of unknown, so it can be cast automatically to other types).
StringBinaryDumper.register(str, ctx)
+ StringDumper.register(str, ctx)
TextLoader.register(INVALID_OID, ctx)
TextLoader.register("bpchar", ctx)
TextLoader.register("name", ctx)
TimestamptzLoader.register("timestamptz", ctx)
IntervalLoader.register("interval", ctx)
- JsonDumper.register(Json, ctx)
+ # Currently json binary format is nothing different than text, maybe with
+ # an extra memcopy we can avoid.
JsonBinaryDumper.register(Json, ctx)
- JsonbDumper.register(Jsonb, ctx)
+ JsonDumper.register(Json, ctx)
JsonbBinaryDumper.register(Jsonb, ctx)
+ JsonbDumper.register(Jsonb, ctx)
JsonLoader.register("json", ctx)
JsonbLoader.register("jsonb", ctx)
JsonBinaryLoader.register("json", ctx)
m(spec, g, w)
def get_supported_types(self):
- dumpers = self.conn.adapters._dumpers[Format.as_pq(self.format)]
+ dumpers = self.conn.adapters._dumpers[self.format]
rv = set()
for cls in dumpers.keys():
if isinstance(cls, str):
def test_dump_connection_ctx(conn):
- make_dumper("t").register(MyStr, conn)
make_bin_dumper("b").register(MyStr, conn)
+ make_dumper("t").register(MyStr, conn)
cur = conn.cursor()
cur.execute("select %s", [MyStr("hello")])
def test_dump_cursor_ctx(conn):
- make_dumper("t").register(str, conn)
make_bin_dumper("b").register(str, conn)
+ make_dumper("t").register(str, conn)
cur = conn.cursor()
- make_dumper("tc").register(str, cur)
make_bin_dumper("bc").register(str, cur)
+ make_dumper("tc").register(str, cur)
cur.execute("select %s", [MyStr("hello")])
assert cur.fetchone() == ("hellotc",)
assert t.get_dumper(L, fmt_in)
-def test_string_connection_ctx(conn):
- make_dumper("t").register(str, conn)
- make_bin_dumper("b").register(str, conn)
-
+def test_last_dumper_registered_ctx(conn):
cur = conn.cursor()
- cur.execute("select %s", ["hello"])
- assert cur.fetchone() == ("hellot",) # str prefers text
- cur.execute("select %t", ["hello"])
- assert cur.fetchone() == ("hellot",)
- cur.execute("select %b", ["hello"])
- assert cur.fetchone() == ("hellob",)
+
+ bd = make_bin_dumper("b")
+ bd.register(str, cur)
+ td = make_dumper("t")
+ td.register(str, cur)
+
+ assert cur.execute("select %s", ["hello"]).fetchone()[0] == "hellot"
+ assert cur.execute("select %t", ["hello"]).fetchone()[0] == "hellot"
+ assert cur.execute("select %b", ["hello"]).fetchone()[0] == "hellob"
+
+ bd.register(str, cur)
+ assert cur.execute("select %s", ["hello"]).fetchone()[0] == "hellob"
@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY])
# All the registered adapters
reg_adapters = set()
adapters = (
- psycopg3.global_adapters._dumpers + psycopg3.global_adapters._loaders
+ list(psycopg3.global_adapters._dumpers.values())
+ + psycopg3.global_adapters._loaders
)
- assert len(adapters) == 4
+ assert len(adapters) == 5
for m in adapters:
reg_adapters |= set(m.values())
projects / thirdparty / psycopg.git / commitdiff
