# Copyright (C) 2020 The Psycopg Team
-from typing import Any, Dict, Iterator, Optional, overload
-from typing import Sequence, Tuple, Type, Union, TYPE_CHECKING
+from __future__ import annotations
+
+from typing import Any, Iterator, overload, Sequence, TYPE_CHECKING
from . import sql
from . import errors as e
from ._connection_base import BaseConnection
T = TypeVar("T", bound="TypeInfo")
-RegistryKey: TypeAlias = Union[str, int, Tuple[type, int]]
+RegistryKey: TypeAlias = "str | int | tuple[type, int]"
class TypeInfo:
@overload
@classmethod
def fetch(
- cls: Type[T], conn: "Connection[Any]", name: Union[str, sql.Identifier]
- ) -> Optional[T]: ...
+ cls: type[T], conn: Connection[Any], name: str | sql.Identifier
+ ) -> T | None: ...
@overload
@classmethod
async def fetch(
- cls: Type[T], conn: "AsyncConnection[Any]", name: Union[str, sql.Identifier]
- ) -> Optional[T]: ...
+ cls: type[T], conn: AsyncConnection[Any], name: str | sql.Identifier
+ ) -> T | None: ...
@classmethod
def fetch(
- cls: Type[T], conn: "BaseConnection[Any]", name: Union[str, sql.Identifier]
+ cls: type[T], conn: BaseConnection[Any], name: str | sql.Identifier
) -> Any:
"""Query a system catalog to read information about a type."""
from .connection import Connection
)
@classmethod
- def _fetch(cls: Type[T], conn: "Connection[Any]", name: str) -> Optional[T]:
+ def _fetch(cls: type[T], conn: Connection[Any], name: str) -> T | None:
# This might result in a nested transaction. What we want is to leave
# the function with the connection in the state we found (either idle
# or intrans)
@classmethod
async def _fetch_async(
- cls: Type[T], conn: "AsyncConnection[Any]", name: str
- ) -> Optional[T]:
+ cls: type[T], conn: AsyncConnection[Any], name: str
+ ) -> T | None:
try:
from psycopg import AsyncCursor
@classmethod
def _from_records(
- cls: Type[T], name: str, recs: Sequence[Dict[str, Any]]
- ) -> Optional[T]:
+ cls: type[T], name: str, recs: Sequence[dict[str, Any]]
+ ) -> T | None:
if len(recs) == 1:
return cls(**recs[0])
elif not recs:
else:
raise e.ProgrammingError(f"found {len(recs)} different types named {name}")
- def register(self, context: Optional[AdaptContext] = None) -> None:
+ def register(self, context: AdaptContext | None = None) -> None:
"""
Register the type information, globally or in the specified `!context`.
"""
register_array(self, context)
@classmethod
- def _get_info_query(cls, conn: "BaseConnection[Any]") -> Query:
+ def _get_info_query(cls, conn: BaseConnection[Any]) -> Query:
return sql.SQL(
"""\
SELECT
).format(regtype=cls._to_regtype(conn))
@classmethod
- def _has_to_regtype_function(cls, conn: "BaseConnection[Any]") -> bool:
+ def _has_to_regtype_function(cls, conn: BaseConnection[Any]) -> bool:
# to_regtype() introduced in PostgreSQL 9.4 and CockroachDB 22.2
info = conn.info
if info.vendor == "PostgreSQL":
return False
@classmethod
- def _to_regtype(cls, conn: "BaseConnection[Any]") -> sql.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.
else:
return sql.SQL("%(name)s::regtype")
- def _added(self, registry: "TypesRegistry") -> None:
+ def _added(self, registry: TypesRegistry) -> None:
"""Method called by the `!registry` when the object is added there."""
pass
__module__ = "psycopg.types"
- def __init__(self, template: Optional["TypesRegistry"] = None):
- self._registry: Dict[RegistryKey, TypeInfo]
+ def __init__(self, template: TypesRegistry | None = None):
+ self._registry: dict[RegistryKey, TypeInfo]
# Make a shallow copy: it will become a proper copy if the registry
# is edited.
yield t
@overload
- def __getitem__(self, key: Union[str, int]) -> TypeInfo: ...
+ def __getitem__(self, key: str | int) -> TypeInfo: ...
@overload
- def __getitem__(self, key: Tuple[Type[T], int]) -> T: ...
+ def __getitem__(self, key: tuple[type[T], int]) -> T: ...
def __getitem__(self, key: RegistryKey) -> TypeInfo:
"""
raise KeyError(f"couldn't find the type {key!r} in the types registry")
@overload
- def get(self, key: Union[str, int]) -> Optional[TypeInfo]: ...
+ def get(self, key: str | int) -> TypeInfo | None: ...
@overload
- def get(self, key: Tuple[Type[T], int]) -> Optional[T]: ...
+ def get(self, key: tuple[type[T], int]) -> T | None: ...
- def get(self, key: RegistryKey) -> Optional[TypeInfo]:
+ def get(self, key: RegistryKey) -> TypeInfo | None:
"""
Return info about a type, specified by name or oid
else:
return t.oid
- def get_by_subtype(self, cls: Type[T], subtype: Union[int, str]) -> Optional[T]:
+ def get_by_subtype(self, cls: type[T], subtype: int | str) -> T | None:
"""
Return info about a `TypeInfo` subclass by its element name or oid.