"""
Helper object to transform values between Python and PostgreSQL
+
+Python implementation of the object. Use the `_transformer module to import
+the right implementation (Python or C). The public place where the object
+is exported is `psycopg.adapt` (which we may not use to avoid circular
+dependencies problems).
"""
# Copyright (C) 2020 The Psycopg Team
from typing_extensions import TypeAlias
from . import pq
-from . import postgres
+from . import abc
from . import errors as e
from .abc import Buffer, LoadFunc, AdaptContext, PyFormat, DumperKey, NoneType
from .rows import Row, RowMaker
from ._encodings import pgconn_encoding
if TYPE_CHECKING:
- from .abc import Dumper, Loader
from .adapt import AdaptersMap
from .pq.abc import PGresult
from .connection import BaseConnection
-DumperCache: TypeAlias = Dict[DumperKey, "Dumper"]
-OidDumperCache: TypeAlias = Dict[int, "Dumper"]
-LoaderCache: TypeAlias = Dict[int, "Loader"]
+DumperCache: TypeAlias = Dict[DumperKey, abc.Dumper]
+OidDumperCache: TypeAlias = Dict[int, abc.Dumper]
+LoaderCache: TypeAlias = Dict[int, abc.Loader]
TEXT = pq.Format.TEXT
PY_TEXT = PyFormat.TEXT
self._adapters = context.adapters
self._conn = context.connection
else:
+ from . import postgres
+
self._adapters = postgres.adapters
self._conn = None
# mapping fmt, oid -> Loader instance
self._loaders: Tuple[LoaderCache, LoaderCache] = ({}, {})
- self._row_dumpers: Optional[List["Dumper"]] = None
+ self._row_dumpers: Optional[List[abc.Dumper]] = None
# sequence of load functions from value to python
# the length of the result columns
rv = bytes(rv)
return rv
- def get_dumper(self, obj: Any, format: PyFormat) -> "Dumper":
+ def get_dumper(self, obj: Any, format: PyFormat) -> abc.Dumper:
"""
Return a Dumper instance to dump `!obj`.
"""
return rv
- def get_dumper_by_oid(self, oid: int, format: pq.Format) -> "Dumper":
+ def get_dumper_by_oid(self, oid: int, format: pq.Format) -> abc.Dumper:
"""
Return a Dumper to dump an object to the type with given oid.
"""
for i, val in enumerate(record)
)
- def get_loader(self, oid: int, format: pq.Format) -> "Loader":
+ def get_loader(self, oid: int, format: pq.Format) -> abc.Loader:
try:
return self._loaders[format][oid]
except KeyError:
--- /dev/null
+"""
+Helper object to transform values between Python and PostgreSQL
+
+This module exports the requested implementation to the rest of the package.
+"""
+
+# Copyright (C) 2023 The Psycopg Team
+
+from typing import Type
+
+from . import abc
+from ._cmodule import _psycopg
+
+Transformer: Type[abc.Transformer]
+
+if _psycopg:
+ Transformer = _psycopg.Transformer
+else:
+ from . import _py_transformer
+
+ Transformer = _py_transformer.Transformer
from typing import Sequence, Tuple, Type, TypeVar, Union, TYPE_CHECKING
from typing_extensions import TypeAlias
+from . import sql
from . import errors as e
from .abc import AdaptContext, Query
from .rows import dict_row
if TYPE_CHECKING:
from .connection import BaseConnection, Connection
from .connection_async import AsyncConnection
- from .sql import Identifier, SQL
T = TypeVar("T", bound="TypeInfo")
RegistryKey: TypeAlias = Union[str, int, Tuple[type, int]]
@overload
@classmethod
def fetch(
- cls: Type[T], conn: "Connection[Any]", name: Union[str, "Identifier"]
+ cls: Type[T], conn: "Connection[Any]", name: Union[str, sql.Identifier]
) -> Optional[T]:
...
@overload
@classmethod
async def fetch(
- cls: Type[T], conn: "AsyncConnection[Any]", name: Union[str, "Identifier"]
+ cls: Type[T], conn: "AsyncConnection[Any]", name: Union[str, sql.Identifier]
) -> Optional[T]:
...
@classmethod
def fetch(
- cls: Type[T], conn: "BaseConnection[Any]", name: Union[str, "Identifier"]
+ cls: Type[T], conn: "BaseConnection[Any]", name: Union[str, sql.Identifier]
) -> Any:
"""Query a system catalog to read information about a type."""
- from .sql import Composable
from .connection import Connection
from .connection_async import AsyncConnection
- if isinstance(name, Composable):
+ if isinstance(name, sql.Composable):
name = name.as_string(conn)
if isinstance(conn, Connection):
@classmethod
def _get_info_query(cls, conn: "BaseConnection[Any]") -> Query:
- from .sql import SQL
-
- return SQL(
+ return sql.SQL(
"""\
SELECT
typname AS name, oid, typarray AS array_oid,
return False
@classmethod
- def _to_regtype(cls, conn: "BaseConnection[Any]") -> "SQL":
+ def _to_regtype(cls, conn: "BaseConnection[Any]") -> sql.SQL:
# `to_regtype()` returns the type oid or NULL, unlike the :: operator,
# which returns the type or raises an exception, which requires
# a transaction rollback and leaves traces in the server logs.
- from .sql import SQL
-
if cls._has_to_regtype_function(conn):
- return SQL("to_regtype(%(name)s)")
+ return sql.SQL("to_regtype(%(name)s)")
else:
- return SQL("%(name)s::regtype")
+ return sql.SQL("%(name)s::regtype")
def _added(self, registry: "TypesRegistry") -> None:
"""Method called by the `!registry` when the object is added there."""
# Copyright (C) 2020 The Psycopg Team
from abc import ABC, abstractmethod
-from typing import Any, Optional, Type, TYPE_CHECKING
+from typing import Any, Optional, TYPE_CHECKING
from . import pq, abc
-from . import _adapters_map
+
+# Objects exported here
from ._enums import PyFormat as PyFormat
-from ._cmodule import _psycopg
+from ._transformer import Transformer as Transformer
+from ._adapters_map import AdaptersMap as AdaptersMap # noqa: F401
if TYPE_CHECKING:
from .connection import BaseConnection
-AdaptersMap = _adapters_map.AdaptersMap
Buffer = abc.Buffer
ORD_BS = ord("\\")
...
-Transformer: Type["abc.Transformer"]
-
-# Override it with fast object if available
-if _psycopg:
- Transformer = _psycopg.Transformer
-else:
- from . import _transform
-
- Transformer = _transform.Transformer
-
-
class RecursiveDumper(Dumper):
"""Dumper with a transformer to help dumping recursive types."""
from .pq import Escaping
from .abc import AdaptContext
-from .adapt import Transformer, PyFormat
+from ._enums import PyFormat
from ._compat import LiteralString
from ._encodings import conn_encoding
+from ._transformer import Transformer
def quote(obj: Any, context: Optional[AdaptContext] = None) -> str:
projects / thirdparty / psycopg.git / commitdiff
