Use the format parameter on register(): documented and made public.
Dumpers and loaders can be registered on different scopes: globally, per
`~psycopg3.Connection`, per `~psycopg3.Cursor`, so that adaptation rules can
be customised for specific needs within the same application: in order to do
-so you can use the *context* parameter of `~Dumper.register()` and similar
-methods.
+so you can use the *context* parameter of `Dumper.register()` and
+`Loader.register()`.
-Dumpers and loaders might need to handle data in text and binary format,
-according to how they are registered (e.g. with `~Dumper.register()` or
-`~Dumper.register_binary()`). For most types the format is different so there
-will have to be two different classes.
+.. note::
+
+ `!register()` is a class method on the base class, so if you
+ subclass `!Dumper` or `!Loader` you should call the ``.register()`` on the
+ class you created.
+
+If the data type has also a distinct binary format which you may want to use
+from psycopg (as documented in :ref:`binary-data`) you may want to implement
+binary loaders and dumpers, whose only difference from text dumper is that
+they must be registered using ``format=Format.BINARY`` in `Dumper.register()`
+and `Loader.register()`.
.. admonition:: TODO
.. automethod:: quote
- By default will return the `dump()` value quoted and sanitised, so
- that the result can be used to build a SQL string. For instance, the
- method will be used by `~psycopg3.sql.Literal` to convert a value
- client-side.
+ By default return the `dump()` value quoted and sanitised, so
+ that the result can be used to build a SQL string. This works well
+ for most types and you won't likely have to implement this method in a
+ subclass.
+
+ .. tip::
+
+ This method will be used by `~psycopg3.sql.Literal` to convert a
+ value client-side.
This method only makes sense for text dumpers; the result of calling
it on a binary dumper is undefined. It might scratch your car, or burn
.. autoattribute:: oid
:annotation: int
- .. automethod:: register(src, context=None)
+ .. automethod:: register(src, context=None, format=Format.TEXT)
+
+ You should call this method on the `Dumper` subclass you create,
+ passing the Python type you want to dump as *src*.
:param src: The type to manage.
:type src: `!type` or `!str`
:param context: Where the dumper should be used. If `!None` the dumper
will be used globally.
:type context: `~psycopg3.Connection`, `~psycopg3.Cursor`, or `Transformer`
+ :param format: Register the dumper for text or binary adaptation
+ :type format: `~psycopg3.pq.Format`
If *src* is specified as string it will be lazy-loaded, so that it
will be possible to register it without importing it before. In this
case it should be the fully qualified name of the object (e.g.
``"uuid.UUID"``).
- .. automethod:: register_binary(src, context=None)
-
- In order to convert a value in binary you can use a ``%b`` placeholder
- in the query instead of ``%s``.
-
- Parameters as the same as in `register()`.
-
.. autoclass:: Loader(oid, context=None)
.. automethod:: load
- .. automethod:: register(oid, context=None)
+ .. automethod:: register(oid, context=None, format=Format.TEXT)
+
+ You should call this method on the `Loader` subclass you create,
+ passing the OID of the type you want to load as *oid* parameter.
:param oid: The PostgreSQL OID to manage.
:type oid: `!int`
:param context: Where the loader should be used. If `!None` the loader
will be used globally.
:type context: `~psycopg3.Connection`, `~psycopg3.Cursor`, or `Transformer`
-
- .. automethod:: register_binary(oid, context=None)
-
- Parameters as the same as in `register()`.
+ :param format: Register the loader for text or binary adaptation
+ :type format: `~psycopg3.pq.Format`
.. autoclass:: Transformer(context=None)
where = context.dumpers if context else Dumper.globals
where[src, format] = cls
- @classmethod
- def register_binary(
- cls, src: Union[type, str], context: AdaptContext = None
- ) -> None:
- """
- Configure *context* to use this dumper for binary format conversion.
- """
- cls.register(src, context, format=Format.BINARY)
-
@classmethod
def text(cls, src: Union[type, str]) -> Callable[[DumperType], DumperType]:
def text_(dumper: DumperType) -> DumperType:
cls, src: Union[type, str]
) -> Callable[[DumperType], DumperType]:
def binary_(dumper: DumperType) -> DumperType:
- dumper.register_binary(src)
+ dumper.register(src, format=Format.BINARY)
return dumper
return binary_
where = context.loaders if context else Loader.globals
where[oid, format] = cls
- @classmethod
- def register_binary(cls, oid: int, context: AdaptContext = None) -> None:
- """
- Configure *context* to use this loader to convert binary values.
- """
- cls.register(oid, context, format=Format.BINARY)
-
@classmethod
def text(cls, oid: int) -> Callable[[LoaderType], LoaderType]:
def text_(loader: LoaderType) -> LoaderType:
@classmethod
def binary(cls, oid: int) -> Callable[[LoaderType], LoaderType]:
def binary_(loader: LoaderType) -> LoaderType:
- loader.register_binary(oid)
+ loader.register(oid, format=Format.BINARY)
return loader
return binary_
where = context.dumpers if context else Dumper.globals
where[src, format] = cls
- @classmethod
- def register_binary(
- cls, src: Union[type, str], context: AdaptContext = None
- ) -> None:
- cls.register(src, context, format=Format.BINARY)
-
cdef class CLoader:
cdef impl.Oid _oid
where = context.loaders if context else Loader.globals
where[oid, format] = cls
- @classmethod
- def register_binary(
- cls, oid: int, context: AdaptContext = None
- ) -> None:
- cls.register(oid, context, format=Format.BINARY)
-
cdef _connection_from_context(object context):
from psycopg3.adapt import _connection_from_context
logger.debug("registering optimised numeric c adapters")
IntDumper.register(int)
- IntBinaryDumper.register_binary(int)
+ IntBinaryDumper.register(int, format=Format.BINARY)
IntLoader.register(oids.INT2_OID)
IntLoader.register(oids.INT4_OID)
FloatLoader.register(oids.FLOAT4_OID)
FloatLoader.register(oids.FLOAT8_OID)
- Int2BinaryLoader.register_binary(oids.INT2_OID)
- Int4BinaryLoader.register_binary(oids.INT4_OID)
- Int8BinaryLoader.register_binary(oids.INT8_OID)
- OidBinaryLoader.register_binary(oids.OID_OID)
- Float4BinaryLoader.register_binary(oids.FLOAT4_OID)
- Float8BinaryLoader.register_binary(oids.FLOAT8_OID)
+ Int2BinaryLoader.register(oids.INT2_OID, format=Format.BINARY)
+ Int4BinaryLoader.register(oids.INT4_OID, format=Format.BINARY)
+ Int8BinaryLoader.register(oids.INT8_OID, format=Format.BINARY)
+ OidBinaryLoader.register(oids.OID_OID, format=Format.BINARY)
+ Float4BinaryLoader.register(oids.FLOAT4_OID, format=Format.BINARY)
+ Float8BinaryLoader.register(oids.FLOAT8_OID, format=Format.BINARY)
# Copyright (C) 2020 The Psycopg Team
+from psycopg3.pq import Format
from psycopg3_c cimport oids
logger.debug("registering optimised singletons c adapters")
BoolDumper.register(bool)
- BoolBinaryDumper.register_binary(bool)
+ BoolBinaryDumper.register(bool, format=Format.BINARY)
BoolLoader.register(oids.BOOL_OID)
- BoolBinaryLoader.register_binary(oids.BOOL_OID)
+ BoolBinaryLoader.register(oids.BOOL_OID, format=Format.BINARY)
logger.debug("registering optimised text c adapters")
StringDumper.register(str)
- StringBinaryDumper.register_binary(str)
+ StringBinaryDumper.register(str, format=Format.BINARY)
TextLoader.register(oids.INVALID_OID)
TextLoader.register(oids.TEXT_OID)
- TextLoader.register_binary(oids.TEXT_OID)
+ TextLoader.register(oids.TEXT_OID, format=Format.BINARY)
TextLoader.register(oids.VARCHAR_OID)
- TextLoader.register_binary(oids.VARCHAR_OID)
+ TextLoader.register(oids.VARCHAR_OID, format=Format.BINARY)
ByteaLoader.register(oids.BYTEA_OID)
- ByteaBinaryLoader.register_binary(oids.BYTEA_OID)
+ ByteaBinaryLoader.register(oids.BYTEA_OID, format=Format.BINARY)
def test_dump_connection_ctx(conn):
make_dumper("t").register(str, conn)
- make_dumper("b").register_binary(str, conn)
+ make_dumper("b").register(str, conn, format=Format.BINARY)
cur = conn.cursor()
cur.execute("select %s, %b", ["hello", "world"])
def test_dump_cursor_ctx(conn):
make_dumper("t").register(str, conn)
- make_dumper("b").register_binary(str, conn)
+ make_dumper("b").register(str, conn, format=Format.BINARY)
cur = conn.cursor()
make_dumper("tc").register(str, cur)
- make_dumper("bc").register_binary(str, cur)
+ make_dumper("bc").register(str, cur, format=Format.BINARY)
cur.execute("select %s, %b", ["hello", "world"])
assert cur.fetchone() == ("hellotc", "worldbc")
def test_load_connection_ctx(conn):
make_loader("t").register(TEXT_OID, conn)
- make_loader("b").register_binary(TEXT_OID, conn)
+ make_loader("b").register(TEXT_OID, conn, format=Format.BINARY)
r = conn.cursor().execute("select 'hello'::text").fetchone()
assert r == ("hellot",)
def test_load_cursor_ctx(conn):
make_loader("t").register(TEXT_OID, conn)
- make_loader("b").register_binary(TEXT_OID, conn)
+ make_loader("b").register(TEXT_OID, conn, format=Format.BINARY)
cur = conn.cursor()
make_loader("tc").register(TEXT_OID, cur)
- make_loader("bc").register_binary(TEXT_OID, cur)
+ make_loader("bc").register(TEXT_OID, cur, format=Format.BINARY)
r = cur.execute("select 'hello'::text").fetchone()
assert r == ("hellotc",)