.. autoclass:: AdaptContext
:members:
+
+ .. seealso:: :ref:`adaptation` for an explanation about how contexts are
+ connected.
.. autoclass:: AdaptersMap
+ .. seealso:: :ref:`adaptation` for an explanation about how contexts are
+ connected.
+
.. automethod:: register_dumper
.. automethod:: register_loader
database as a list of the base type.
-.. autoclass:: TypesRegistry
-
-
The following `!TypeInfo` subclasses allow to fetch more specialised
-information from certain class of PostgreSQL types and to create more
-specialised adapters configurations.
-
+information from certain class of PostgreSQL types.
.. autoclass:: psycopg.types.composite.CompositeInfo
- .. automethod:: register
-
- Using `!CompositeInfo.register()` will also register a specialised
- loader to fetch the composite type as a Python named tuple, or a
- custom object if *factory* is specified.
-
-
.. autoclass:: psycopg.types.range.RangeInfo
- .. automethod:: register
- Using `!RangeInfo.register()` will also register a specialised loaders
- and dumpers. For instance, if you create a PostgreSQL range on the
- type :sql:`inet`, loading these object with the database will use the
- loader for the :sql:`inet` type to parse the range bounds - either the
- builtin ones or any one you might have configured.
+`!TypeInfo` objects are collected in `TypesRegistry` instances, which help type
+information lookup. Every `~psycopg.adapt.AdaptersMap` expose its type map on
+its `~psycopg.adapt.AdaptersMap.types` attribute.
- The type information will also be used by the `Range` dumper so that
- if you dump a `!Range(address1, address2)` object it will use the
- correct oid for your :sql:`inetrange` type.
+.. autoclass:: TypesRegistry
.. _numeric-wrappers:
# Copyright (C) 2020-2021 The Psycopg Team
-from typing import Any, Callable, Dict, Iterator, Optional
+from typing import Any, Dict, Iterator, Optional
from typing import Sequence, Type, TypeVar, Union, TYPE_CHECKING
from . import errors as e
super().__init__(name, oid, array_oid)
self.subtype_oid = subtype_oid
- def register(self, context: Optional[AdaptContext] = None) -> None:
- super().register(context)
-
- from .types.range import register_range
-
- register_range(self, context)
-
_info_query = """\
SELECT t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
r.rngsubtype AS subtype_oid
# Will be set by register() if the `factory` is a type
self.python_type: Optional[type] = None
- def register(
- self,
- context: Optional[AdaptContext] = None,
- factory: Optional[Callable[..., Any]] = None,
- ) -> None:
- super().register(context)
-
- from .types.composite import register_composite
-
- register_composite(self, context, factory)
-
_info_query = """\
SELECT
t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
context: Optional[AdaptContext] = None,
factory: Optional[Callable[..., Any]] = None,
) -> None:
+
+ # Register arrays and type info
+ info.register(context)
+
if not factory:
factory = namedtuple(info.name, info.field_names) # type: ignore
info: TypeInfo, context: Optional[AdaptContext] = None
) -> None:
+ # Register arrays and type info
info.register(context)
adapters = context.adapters if context else postgres.adapters
right subtype. Dumping the range just works, navigating from tye Python
type to the type oid, to the range oid.
"""
+
+ # Register arrays and type info
+ info.register(context)
+
adapters = context.adapters if context else postgres.adapters
# generate and register a customized text loader
from psycopg.sql import Identifier
from psycopg.adapt import PyFormat as Format
from psycopg.postgres import types as builtins
-from psycopg.types.composite import CompositeInfo, TupleDumper
-
+from psycopg.types.composite import CompositeInfo, register_composite
+from psycopg.types.composite import TupleDumper
tests_str = [
("", ()),
"""
)
info = CompositeInfo.fetch(conn, "tmptype")
- info.register(context=conn)
+ register_composite(info, conn)
res = conn.execute("select %s::tmptype", [obj]).fetchone()[0]
assert res == obj
@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
def test_load_composite(conn, testcomp, fmt_out):
info = CompositeInfo.fetch(conn, "testcomp")
- info.register(conn)
+ register_composite(info, conn)
cur = conn.cursor(binary=fmt_out)
res = cur.execute("select row('hello', 10, 20)::testcomp").fetchone()[0]
def __init__(self, *args):
self.foo, self.bar, self.baz = args
- info.register(conn, factory=MyThing)
+ register_composite(info, conn, factory=MyThing)
assert info.python_type is MyThing
cur = conn.cursor(binary=fmt_out)
def test_register_scope(conn, testcomp):
info = CompositeInfo.fetch(conn, "testcomp")
- info.register()
+ register_composite(info)
for fmt in (pq.Format.TEXT, pq.Format.BINARY):
for oid in (info.oid, info.array_oid):
assert postgres.adapters._loaders[fmt].pop(oid)
assert info.python_type not in postgres.adapters._dumpers[Format.BINARY]
cur = conn.cursor()
- info.register(cur)
+ register_composite(info, cur)
for fmt in (pq.Format.TEXT, pq.Format.BINARY):
for oid in (info.oid, info.array_oid):
assert oid not in postgres.adapters._loaders[fmt]
assert oid not in conn.adapters._loaders[fmt]
assert oid in cur.adapters._loaders[fmt]
- info.register(conn)
+ register_composite(info, conn)
for fmt in (pq.Format.TEXT, pq.Format.BINARY):
for oid in (info.oid, info.array_oid):
assert oid not in postgres.adapters._loaders[fmt]
def test_type_dumper_registered(conn, testcomp):
info = CompositeInfo.fetch(conn, "testcomp")
- info.register(conn)
+ register_composite(info, conn)
assert issubclass(info.python_type, tuple)
assert info.python_type.__name__ == "testcomp"
d = conn.adapters.get_dumper(info.python_type, "s")
def fac(*args):
return args + (args[-1],)
- info.register(conn, factory=fac)
+ register_composite(info, conn, factory=fac)
assert info.python_type is None
# but the loader is registered
from psycopg.sql import Identifier
from psycopg.adapt import PyFormat as Format
from psycopg.types import range as range_module
-from psycopg.types.range import Range, RangeInfo
+from psycopg.types.range import Range, RangeInfo, register_range
type2sub = {
def test_dump_custom_empty(conn, testrange):
info = RangeInfo.fetch(conn, "testrange")
- info.register(conn)
+ register_range(info, conn)
r = Range(empty=True)
cur = conn.execute("select 'empty'::testrange = %s", (r,))
def test_dump_quoting(conn, testrange):
info = RangeInfo.fetch(conn, "testrange")
- info.register(conn)
+ register_range(info, conn)
cur = conn.cursor()
for i in range(1, 254):
cur.execute(
@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
def test_load_custom_empty(conn, testrange, fmt_out):
info = RangeInfo.fetch(conn, "testrange")
- info.register(conn)
+ register_range(info, conn)
cur = conn.cursor(binary=fmt_out)
(got,) = cur.execute("select 'empty'::testrange").fetchone()
@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
def test_load_quoting(conn, testrange, fmt_out):
info = RangeInfo.fetch(conn, "testrange")
- info.register(conn)
+ register_range(info, conn)
cur = conn.cursor(binary=fmt_out)
for i in range(1, 254):
cur.execute(
projects / thirdparty / psycopg.git / commitdiff
