from . import errors as e
from .abc import Buffer, LoadFunc, AdaptContext, PyFormat, DumperKey, NoneType
from .rows import Row, RowMaker
-from .postgres import INVALID_OID, TEXT_OID
+from ._oids import INVALID_OID, TEXT_OID
from ._encodings import pgconn_encoding
if TYPE_CHECKING:
"""
# Copyright (C) 2020 The Psycopg Team
+
from enum import Enum
from typing import Any, Dict, Iterator, Optional, overload
from typing import Sequence, Tuple, Type, TypeVar, Union, TYPE_CHECKING
from .._typeinfo import TypeInfo, TypesRegistry
from ..abc import AdaptContext, NoneType
-from ..postgres import TEXT_OID
+from .._oids import TEXT_OID
from .._adapters_map import AdaptersMap
from ..types.enum import EnumDumper, EnumBinaryDumper
from ..types.none import NoneDumper
types.add(t)
-# A few oids used a bit everywhere
-INVALID_OID = 0
-TEXT_OID = types["text"].oid
-TEXT_ARRAY_OID = types["text"].array_oid
-
-
def register_default_adapters(context: AdaptContext) -> None:
from .types import array, bool, composite, datetime, enum, json, multirange
from .. import postgres
from ..abc import AdaptContext, Buffer, Dumper, DumperKey, NoneType, Loader, Transformer
from ..adapt import RecursiveDumper, RecursiveLoader, PyFormat
+from .._oids import TEXT_OID, INVALID_OID
from .._compat import cache, prod
from .._struct import pack_len, unpack_len
from .._cmodule import _psycopg
-from ..postgres import TEXT_OID, INVALID_OID
from .._typeinfo import TypeInfo
_struct_head = struct.Struct("!III") # ndims, hasnull, elem oid
from .. import postgres
from ..abc import AdaptContext, Buffer
from ..adapt import Transformer, PyFormat, RecursiveDumper, Loader
+from .._oids import TEXT_OID
from .._struct import pack_len, unpack_len
-from ..postgres import TEXT_OID
from .._typeinfo import CompositeInfo as CompositeInfo # exported here
from .._encodings import _as_python_identifier
from .. import errors as e
from .. import postgres
from ..abc import Buffer, AdaptContext
+from .._oids import TEXT_OID
from ..adapt import PyFormat, RecursiveDumper, RecursiveLoader
-from ..postgres import TEXT_OID
from .._typeinfo import TypeInfo
_re_escape = re.compile(r'(["\\])')
from ..pq import Format
from ..abc import AdaptContext, Buffer, Dumper, DumperKey
from ..adapt import RecursiveDumper, RecursiveLoader, PyFormat
+from .._oids import INVALID_OID, TEXT_OID
from .._struct import pack_len, unpack_len
-from ..postgres import INVALID_OID, TEXT_OID
from .._typeinfo import MultirangeInfo as MultirangeInfo # exported here
from .range import Range, T, load_range_text, load_range_binary
from ..pq import Format
from ..abc import AdaptContext, Buffer, Dumper, DumperKey
from ..adapt import RecursiveDumper, RecursiveLoader, PyFormat
+from .._oids import INVALID_OID, TEXT_OID
from .._struct import pack_len, unpack_len
-from ..postgres import INVALID_OID, TEXT_OID
from .._typeinfo import RangeInfo as RangeInfo # exported here
RANGE_EMPTY = 0x01 # range is empty
from ..abc import AdaptContext
from ..adapt import Buffer, Dumper, Loader
from ..errors import DataError
+from .._oids import INVALID_OID
from .._encodings import conn_encoding
if TYPE_CHECKING:
adapters.register_dumper(str, StrDumper)
adapters.register_dumper(str, StrDumperUnknown)
- adapters.register_loader(postgres.INVALID_OID, TextLoader)
+ adapters.register_loader(INVALID_OID, TextLoader)
adapters.register_loader("bpchar", TextLoader)
adapters.register_loader("name", TextLoader)
adapters.register_loader("text", TextLoader)
adapters.register_dumper(memoryview, BytesBinaryDumper)
adapters.register_loader("bytea", ByteaLoader)
- adapters.register_loader(postgres.INVALID_OID, ByteaBinaryLoader)
+ adapters.register_loader(INVALID_OID, ByteaBinaryLoader)
adapters.register_loader("bytea", ByteaBinaryLoader)
from psycopg import errors as e
from psycopg.adapt import Transformer, PyFormat, Dumper, Loader
from psycopg._cmodule import _psycopg
-from psycopg.postgres import types as builtins, TEXT_OID
+from psycopg.postgres import types as builtins
from psycopg.types.array import ListDumper, ListBinaryDumper
def test_register_loader_by_oid(conn):
- assert TEXT_OID == 25
+ oid = builtins["text"].oid
+ assert oid == 25
loader = make_loader("x")
- assert conn.adapters.get_loader(TEXT_OID, pq.Format.TEXT) is not loader
- conn.adapters.register_loader(TEXT_OID, loader)
- assert conn.adapters.get_loader(TEXT_OID, pq.Format.TEXT) is loader
+ assert conn.adapters.get_loader(oid, pq.Format.TEXT) is not loader
+ conn.adapters.register_loader(oid, loader)
+ assert conn.adapters.get_loader(oid, pq.Format.TEXT) is loader
def test_register_loader_by_type_name(conn):
+ oid = builtins["text"].oid
+ assert oid == 25
loader = make_loader("x")
- assert conn.adapters.get_loader(TEXT_OID, pq.Format.TEXT) is not loader
+ assert conn.adapters.get_loader(oid, pq.Format.TEXT) is not loader
conn.adapters.register_loader("text", loader)
- assert conn.adapters.get_loader(TEXT_OID, pq.Format.TEXT) is loader
+ assert conn.adapters.get_loader(oid, pq.Format.TEXT) is loader
@pytest.mark.crdb("skip", reason="global adapters don't affect crdb")
"""Create a test dumper appending a suffix to the bytes representation."""
class TestDumper(Dumper):
- oid = TEXT_OID
+ oid = builtins["text"].oid
format = pq.Format.TEXT
def dump(self, s):