# 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 typing_extensions import TypeAlias
pass
-class RangeInfo(TypeInfo):
- """Manage information about a range type."""
-
- __module__ = "psycopg.types.range"
-
- def __init__(
- self,
- name: str,
- oid: int,
- array_oid: int,
- *,
- regtype: str = "",
- subtype_oid: int,
- ):
- super().__init__(name, oid, array_oid, regtype=regtype)
- self.subtype_oid = subtype_oid
-
- @classmethod
- def _get_info_query(cls, conn: "BaseConnection[Any]") -> Query:
- from .sql import SQL
-
- return SQL(
- """\
-SELECT t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
- t.oid::regtype::text AS regtype,
- r.rngsubtype AS subtype_oid
-FROM pg_type t
-JOIN pg_range r ON t.oid = r.rngtypid
-WHERE t.oid = {regtype}
-"""
- ).format(regtype=cls._to_regtype(conn))
-
- def _added(self, registry: "TypesRegistry") -> None:
- # Map ranges subtypes to info
- registry._registry[RangeInfo, self.subtype_oid] = self
-
-
-class MultirangeInfo(TypeInfo):
- """Manage information about a multirange type."""
-
- __module__ = "psycopg.types.multirange"
-
- def __init__(
- self,
- name: str,
- oid: int,
- array_oid: int,
- *,
- regtype: str = "",
- range_oid: int,
- subtype_oid: int,
- ):
- super().__init__(name, oid, array_oid, regtype=regtype)
- self.range_oid = range_oid
- self.subtype_oid = subtype_oid
-
- @classmethod
- def _get_info_query(cls, conn: "BaseConnection[Any]") -> Query:
- from .sql import SQL
-
- if conn.info.server_version < 140000:
- raise e.NotSupportedError(
- "multirange types are only available from PostgreSQL 14"
- )
-
- return SQL(
- """\
-SELECT t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
- t.oid::regtype::text AS regtype,
- r.rngtypid AS range_oid, r.rngsubtype AS subtype_oid
-FROM pg_type t
-JOIN pg_range r ON t.oid = r.rngmultitypid
-WHERE t.oid = {regtype}
-"""
- ).format(regtype=cls._to_regtype(conn))
-
- def _added(self, registry: "TypesRegistry") -> None:
- # Map multiranges ranges and subtypes to info
- registry._registry[MultirangeInfo, self.range_oid] = self
- registry._registry[MultirangeInfo, self.subtype_oid] = self
-
-
-class CompositeInfo(TypeInfo):
- """Manage information about a composite type."""
-
- __module__ = "psycopg.types.composite"
-
- def __init__(
- self,
- name: str,
- oid: int,
- array_oid: int,
- *,
- regtype: str = "",
- field_names: Sequence[str],
- field_types: Sequence[int],
- ):
- super().__init__(name, oid, array_oid, regtype=regtype)
- self.field_names = field_names
- self.field_types = field_types
- # Will be set by register() if the `factory` is a type
- self.python_type: Optional[type] = None
-
- @classmethod
- def _get_info_query(cls, conn: "BaseConnection[Any]") -> Query:
- from .sql import SQL
-
- return SQL(
- """\
-SELECT
- t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
- t.oid::regtype::text AS regtype,
- coalesce(a.fnames, '{{}}') AS field_names,
- coalesce(a.ftypes, '{{}}') AS field_types
-FROM pg_type t
-LEFT JOIN (
- SELECT
- attrelid,
- array_agg(attname) AS fnames,
- array_agg(atttypid) AS ftypes
- FROM (
- SELECT a.attrelid, a.attname, a.atttypid
- FROM pg_attribute a
- JOIN pg_type t ON t.typrelid = a.attrelid
- WHERE t.oid = {regtype}
- AND a.attnum > 0
- AND NOT a.attisdropped
- ORDER BY a.attnum
- ) x
- GROUP BY attrelid
-) a ON a.attrelid = t.typrelid
-WHERE t.oid = {regtype}
-"""
- ).format(regtype=cls._to_regtype(conn))
-
-
-class EnumInfo(TypeInfo):
- """Manage information about an enum type."""
-
- __module__ = "psycopg.types.enum"
-
- def __init__(
- self,
- name: str,
- oid: int,
- array_oid: int,
- labels: Sequence[str],
- ):
- super().__init__(name, oid, array_oid)
- self.labels = labels
- # Will be set by register_enum()
- self.enum: Optional[Type[Enum]] = None
-
- @classmethod
- def _get_info_query(cls, conn: "BaseConnection[Any]") -> Query:
- from .sql import SQL
-
- return SQL(
- """\
-SELECT name, oid, array_oid, array_agg(label) AS labels
-FROM (
- SELECT
- t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
- e.enumlabel AS label
- FROM pg_type t
- LEFT JOIN pg_enum e
- ON e.enumtypid = t.oid
- WHERE t.oid = {regtype}
- ORDER BY e.enumsortorder
-) x
-GROUP BY name, oid, array_oid
-"""
- ).format(regtype=cls._to_regtype(conn))
-
-
class TypesRegistry:
"""
Container for the information about types in a database.
# Copyright (C) 2020 The Psycopg Team
-from ._typeinfo import TypeInfo, RangeInfo, MultirangeInfo, TypesRegistry
+from ._typeinfo import TypeInfo, TypesRegistry
from .abc import AdaptContext
from ._adapters_map import AdaptersMap
def register_default_types(types: TypesRegistry) -> None:
+ from .types.range import RangeInfo
+ from .types.multirange import MultirangeInfo
+
# Use tools/update_oids.py to update this data.
for t in [
TypeInfo('"char"', 18, 1002),
import struct
from collections import namedtuple
from typing import Any, Callable, cast, Iterator, List, Optional
-from typing import Sequence, Tuple, Type
+from typing import Sequence, Tuple, Type, TYPE_CHECKING
from .. import pq
+from .. import sql
from .. import postgres
-from ..abc import AdaptContext, Buffer
+from ..abc import AdaptContext, Buffer, Query
from ..adapt import Transformer, PyFormat, RecursiveDumper, Loader
from .._oids import TEXT_OID
from .._struct import pack_len, unpack_len
-from .._typeinfo import CompositeInfo as CompositeInfo # exported here
+from .._typeinfo import TypeInfo
from .._encodings import _as_python_identifier
+if TYPE_CHECKING:
+ from ..connection import BaseConnection
+
_struct_oidlen = struct.Struct("!Ii")
_pack_oidlen = cast(Callable[[int, int], bytes], _struct_oidlen.pack)
_unpack_oidlen = cast(
)
+class CompositeInfo(TypeInfo):
+ """Manage information about a composite type."""
+
+ def __init__(
+ self,
+ name: str,
+ oid: int,
+ array_oid: int,
+ *,
+ regtype: str = "",
+ field_names: Sequence[str],
+ field_types: Sequence[int],
+ ):
+ super().__init__(name, oid, array_oid, regtype=regtype)
+ self.field_names = field_names
+ self.field_types = field_types
+ # Will be set by register() if the `factory` is a type
+ self.python_type: Optional[type] = None
+
+ @classmethod
+ def _get_info_query(cls, conn: "BaseConnection[Any]") -> Query:
+ return sql.SQL(
+ """\
+SELECT
+ t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
+ t.oid::regtype::text AS regtype,
+ coalesce(a.fnames, '{{}}') AS field_names,
+ coalesce(a.ftypes, '{{}}') AS field_types
+FROM pg_type t
+LEFT JOIN (
+ SELECT
+ attrelid,
+ array_agg(attname) AS fnames,
+ array_agg(atttypid) AS ftypes
+ FROM (
+ SELECT a.attrelid, a.attname, a.atttypid
+ FROM pg_attribute a
+ JOIN pg_type t ON t.typrelid = a.attrelid
+ WHERE t.oid = {regtype}
+ AND a.attnum > 0
+ AND NOT a.attisdropped
+ ORDER BY a.attnum
+ ) x
+ GROUP BY attrelid
+) a ON a.attrelid = t.typrelid
+WHERE t.oid = {regtype}
+"""
+ ).format(regtype=cls._to_regtype(conn))
+
+
class SequenceDumper(RecursiveDumper):
def _dump_sequence(
self, obj: Sequence[Any], start: bytes, end: bytes, sep: bytes
"""
from enum import Enum
from typing import Any, Dict, Generic, Optional, Mapping, Sequence
-from typing import Tuple, Type, TypeVar, Union, cast
+from typing import Tuple, Type, TypeVar, Union, cast, TYPE_CHECKING
from typing_extensions import TypeAlias
+from .. import sql
from .. import postgres
from .. import errors as e
from ..pq import Format
-from ..abc import AdaptContext
+from ..abc import AdaptContext, Query
from ..adapt import Buffer, Dumper, Loader
from .._encodings import conn_encoding
-from .._typeinfo import EnumInfo as EnumInfo # exported here
+from .._typeinfo import TypeInfo
+
+if TYPE_CHECKING:
+ from ..connection import BaseConnection
E = TypeVar("E", bound=Enum)
EnumMapping: TypeAlias = Union[Mapping[E, str], Sequence[Tuple[E, str]], None]
+class EnumInfo(TypeInfo):
+ """Manage information about an enum type."""
+
+ def __init__(
+ self,
+ name: str,
+ oid: int,
+ array_oid: int,
+ labels: Sequence[str],
+ ):
+ super().__init__(name, oid, array_oid)
+ self.labels = labels
+ # Will be set by register_enum()
+ self.enum: Optional[Type[Enum]] = None
+
+ @classmethod
+ def _get_info_query(cls, conn: "BaseConnection[Any]") -> Query:
+ return sql.SQL(
+ """\
+SELECT name, oid, array_oid, array_agg(label) AS labels
+FROM (
+ SELECT
+ t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
+ e.enumlabel AS label
+ FROM pg_type t
+ LEFT JOIN pg_enum e
+ ON e.enumtypid = t.oid
+ WHERE t.oid = {regtype}
+ ORDER BY e.enumsortorder
+) x
+GROUP BY name, oid, array_oid
+"""
+ ).format(regtype=cls._to_regtype(conn))
+
+
class _BaseEnumLoader(Loader, Generic[E]):
"""
Loader for a specific Enum class
# Copyright (C) 2021 The Psycopg Team
from decimal import Decimal
-from typing import Any, Generic, List, Iterable
-from typing import MutableSequence, Optional, Type, Union, overload
+from typing import Any, Generic, List, Iterable, MutableSequence
+from typing import Optional, Type, Union, overload, TYPE_CHECKING
from datetime import date, datetime
+from .. import sql
from .. import _oids
from .. import errors as e
from .. import postgres
from ..pq import Format
-from ..abc import AdaptContext, Buffer, Dumper, DumperKey
+from ..abc import AdaptContext, Buffer, Dumper, DumperKey, Query
from ..adapt import RecursiveDumper, RecursiveLoader, PyFormat
from .._oids import INVALID_OID, TEXT_OID
from .._struct import pack_len, unpack_len
-from .._typeinfo import MultirangeInfo as MultirangeInfo # exported here
+from .._typeinfo import TypeInfo, TypesRegistry
from .range import Range, T, load_range_text, load_range_binary
from .range import dump_range_text, dump_range_binary, fail_dump
+if TYPE_CHECKING:
+ from ..connection import BaseConnection
+
+
+class MultirangeInfo(TypeInfo):
+ """Manage information about a multirange type."""
+
+ def __init__(
+ self,
+ name: str,
+ oid: int,
+ array_oid: int,
+ *,
+ regtype: str = "",
+ range_oid: int,
+ subtype_oid: int,
+ ):
+ super().__init__(name, oid, array_oid, regtype=regtype)
+ self.range_oid = range_oid
+ self.subtype_oid = subtype_oid
+
+ @classmethod
+ def _get_info_query(cls, conn: "BaseConnection[Any]") -> Query:
+ if conn.info.server_version < 140000:
+ raise e.NotSupportedError(
+ "multirange types are only available from PostgreSQL 14"
+ )
+ return sql.SQL(
+ """\
+SELECT t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
+ t.oid::regtype::text AS regtype,
+ r.rngtypid AS range_oid, r.rngsubtype AS subtype_oid
+FROM pg_type t
+JOIN pg_range r ON t.oid = r.rngmultitypid
+WHERE t.oid = {regtype}
+"""
+ ).format(regtype=cls._to_regtype(conn))
+
+ def _added(self, registry: "TypesRegistry") -> None:
+ # Map multiranges ranges and subtypes to info
+ registry._registry[MultirangeInfo, self.range_oid] = self
+ registry._registry[MultirangeInfo, self.subtype_oid] = self
+
class Multirange(MutableSequence[Range[T]]):
"""Python representation for a PostgreSQL multirange type.
import re
from typing import Any, Callable, Dict, Generic, List, Optional, TypeVar, Type, Tuple
-from typing import cast
+from typing import cast, TYPE_CHECKING
from decimal import Decimal
from datetime import date, datetime
+from .. import sql
from .. import _oids
from .. import errors as e
from .. import postgres
from ..pq import Format
-from ..abc import AdaptContext, Buffer, Dumper, DumperKey
+from ..abc import AdaptContext, Buffer, Dumper, DumperKey, Query
from ..adapt import RecursiveDumper, RecursiveLoader, PyFormat
from .._oids import INVALID_OID, TEXT_OID
from .._struct import pack_len, unpack_len
-from .._typeinfo import RangeInfo as RangeInfo # exported here
+from .._typeinfo import TypeInfo, TypesRegistry
+
+if TYPE_CHECKING:
+ from ..connection import BaseConnection
RANGE_EMPTY = 0x01 # range is empty
RANGE_LB_INC = 0x02 # lower bound is inclusive
T = TypeVar("T")
+class RangeInfo(TypeInfo):
+ """Manage information about a range type."""
+
+ def __init__(
+ self,
+ name: str,
+ oid: int,
+ array_oid: int,
+ *,
+ regtype: str = "",
+ subtype_oid: int,
+ ):
+ super().__init__(name, oid, array_oid, regtype=regtype)
+ self.subtype_oid = subtype_oid
+
+ @classmethod
+ def _get_info_query(cls, conn: "BaseConnection[Any]") -> Query:
+ return sql.SQL(
+ """\
+SELECT t.typname AS name, t.oid AS oid, t.typarray AS array_oid,
+ t.oid::regtype::text AS regtype,
+ r.rngsubtype AS subtype_oid
+FROM pg_type t
+JOIN pg_range r ON t.oid = r.rngtypid
+WHERE t.oid = {regtype}
+"""
+ ).format(regtype=cls._to_regtype(conn))
+
+ def _added(self, registry: TypesRegistry) -> None:
+ # Map ranges subtypes to info
+ registry._registry[RangeInfo, self.subtype_oid] = self
+
+
class Range(Generic[T]):
"""Python representation for a PostgreSQL range type.
projects / thirdparty / psycopg.git / commitdiff
