from ._compat import Protocol, TypeAlias, LiteralString
if TYPE_CHECKING:
- from .sql import Composable
+ from . import sql
from .rows import Row, RowMaker
from .pq.abc import PGresult
from .waiting import Wait, Ready
# An object implementing the buffer protocol
Buffer: TypeAlias = Union[bytes, bytearray, memoryview]
-Query: TypeAlias = Union[LiteralString, bytes, "Composable"]
+Query: TypeAlias = Union[LiteralString, bytes, "sql.SQL", "sql.Composed"]
Params: TypeAlias = Union[Sequence[Any], Mapping[str, Any]]
ConnectionType = TypeVar("ConnectionType", bound="BaseConnection[Any]")
PipelineCommand: TypeAlias = Callable[[], None]
)
yield from self._conn._exec_command(query)
- def _make_declare_statement(self, query: Query) -> sql.Composable:
+ def _make_declare_statement(self, query: Query) -> sql.Composed:
if isinstance(query, bytes):
query = query.decode(self._encoding)