types.add(self)
if self.array_oid:
- from .types.array import register_adapters
+ from .types.array import register_array
- register_adapters(self, context)
+ register_array(self, context)
_info_query = """\
SELECT
def register(self, context: Optional[AdaptContext] = None) -> None:
super().register(context)
- from .types.range import register_adapters
+ from .types.range import register_range
- register_adapters(self, context)
+ register_range(self, context)
_info_query = """\
SELECT t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
) -> None:
super().register(context)
- from .types.composite import register_adapters
+ from .types.composite import register_composite
- register_adapters(self, context, factory)
+ register_composite(self, context, factory)
_info_query = """\
SELECT
return agg(dims)
-def register_adapters(
+def register_array(
info: TypeInfo, context: Optional[AdaptContext] = None
) -> None:
adapters = context.adapters if context else postgres.adapters
return type(self).factory(*r)
-def register_adapters(
+def register_composite(
info: CompositeInfo,
context: Optional[AdaptContext] = None,
factory: Optional[Callable[..., Any]] = None,
return rv
-def register_adapters(
+def register_hstore(
info: TypeInfo, context: Optional[AdaptContext] = None
) -> None:
_int2parens = {ord(c): c for c in "[]()"}
-def register_adapters(
+def register_range(
info: RangeInfo, context: Optional[AdaptContext] = None
) -> None:
+ """
+ Register custom range adapters on a context.
+
+ Just register loaders associated to the range oid, loading bounds of the
+ right subtype. Dumping the range just works, navigating from tye Python
+ type to the type oid, to the range oid.
+ """
adapters = context.adapters if context else postgres.adapters
# generate and register a customized text loader
from psycopg.pq import Format
from psycopg.adapt import PyFormat as PgFormat
from psycopg.types import TypeInfo
-from psycopg.types.hstore import register_adapters as register_hstore
+from psycopg.types.hstore import register_hstore
from psycopg.types.numeric import Int4
from .utils import gc_collect
from psycopg.pq import Format
from psycopg.types import TypeInfo
from psycopg.adapt import PyFormat as PgFormat
-from psycopg.types.hstore import register_adapters as register_hstore
+from psycopg.types.hstore import register_hstore
from .utils import gc_collect
from .test_copy import sample_text, sample_binary, sample_binary_rows # noqa
import psycopg
from psycopg.types import TypeInfo
-from psycopg.types.hstore import HstoreLoader, register_adapters
+from psycopg.types.hstore import HstoreLoader, register_hstore
@pytest.mark.parametrize(
def test_register_conn(hstore, conn):
info = TypeInfo.fetch(conn, "hstore")
- register_adapters(info, conn)
+ register_hstore(info, conn)
assert conn.adapters.types[info.oid].name == "hstore"
cur = conn.execute("select null::hstore, ''::hstore, 'a => b'::hstore")
def test_register_curs(hstore, conn):
info = TypeInfo.fetch(conn, "hstore")
cur = conn.cursor()
- register_adapters(info, cur)
+ register_hstore(info, cur)
assert conn.adapters.types.get(info.oid) is None
assert cur.adapters.types[info.oid].name == "hstore"
def test_register_globally(hstore, dsn, svcconn, global_adapters):
info = TypeInfo.fetch(svcconn, "hstore")
- register_adapters(info)
+ register_hstore(info)
assert psycopg.adapters.types[info.oid].name == "hstore"
assert svcconn.adapters.types.get(info.oid) is None
@pytest.mark.parametrize("d", samp)
def test_roundtrip(hstore, conn, d):
- register_adapters(TypeInfo.fetch(conn, "hstore"), conn)
+ register_hstore(TypeInfo.fetch(conn, "hstore"), conn)
d1 = conn.execute("select %s", [d]).fetchone()[0]
assert d == d1
def test_roundtrip_array(hstore, conn):
- register_adapters(TypeInfo.fetch(conn, "hstore"), conn)
+ register_hstore(TypeInfo.fetch(conn, "hstore"), conn)
samp1 = conn.execute("select %s", (samp,)).fetchone()[0]
assert samp1 == samp