import functools
import re
from collections import namedtuple
-from typing import Any, Callable, Dict, Sequence, Tuple, Type
+from typing import Any, Callable, Dict, Sequence, Type, NamedTuple
+from typing import TYPE_CHECKING
-from .cursor import BaseCursor
-from .proto import ConnectionType
+from . import errors as e
+
+if TYPE_CHECKING:
+ from .cursor import BaseCursor
+
+
+def tuple_row(cursor: "BaseCursor[Any]") -> None:
+ """Row factory to represent rows as simple tuples.
+
+ This is the default factory.
+ """
+ # Implementation detail: just return None instead of a callable because
+ # the Transformer knows how to use this value.
+ return None
def dict_row(
- cursor: BaseCursor[ConnectionType],
+ cursor: "BaseCursor[Any]",
) -> Callable[[Sequence[Any]], Dict[str, Any]]:
- """Row factory to represent rows as dicts."""
+ """Row factory to represent rows as dicts.
+
+ Note that this is not compatible with the DBAPI, which expects the records
+ to be sequences.
+ """
def make_row(values: Sequence[Any]) -> Dict[str, Any]:
- assert cursor.description
- titles = (c.name for c in cursor.description)
+ desc = cursor.description
+ if desc is None:
+ raise e.InterfaceError("The cursor doesn't have a result")
+ titles = (c.name for c in desc)
return dict(zip(titles, values))
return make_row
def namedtuple_row(
- cursor: BaseCursor[ConnectionType],
-) -> Callable[[Sequence[Any]], Tuple[Any, ...]]:
+ cursor: "BaseCursor[Any]",
+) -> Callable[[Sequence[Any]], NamedTuple]:
"""Row factory to represent rows as `~collections.namedtuple`."""
- def make_row(values: Sequence[Any]) -> Tuple[Any, ...]:
- assert cursor.description
- key = tuple(c.name for c in cursor.description)
+ def make_row(values: Sequence[Any]) -> NamedTuple:
+ desc = cursor.description
+ if desc is None:
+ raise e.InterfaceError("The cursor doesn't have a result")
+ key = tuple(c.name for c in desc)
nt = _make_nt(key)
- rv = nt._make(values) # type: ignore[attr-defined]
- return rv # type: ignore[no-any-return]
+ rv = nt._make(values)
+ return rv
return make_row
@functools.lru_cache(512)
-def _make_nt(key: Sequence[str]) -> Type[Tuple[Any, ...]]:
+def _make_nt(key: Sequence[str]) -> Type[NamedTuple]:
fields = []
for s in key:
s = _re_clean.sub("_", s)
if s[0] == "_" or "0" <= s[0] <= "9":
s = "f" + s
fields.append(s)
- return namedtuple("Row", fields)
+ return namedtuple("Row", fields) # type: ignore[return-value]