1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
"""Database tools.
"""
import io
from typing import (
IO, Any, Iterable, Mapping, Optional, Sequence, Tuple, Type, Union,
)
try:
from typing import Protocol
except ImportError:
Protocol = object # type: ignore
ExecuteParams = Union[Sequence[Any], Mapping[str, Any]]
class DictRow(Protocol):
"""Allow both key and index-based access.
Both Psycopg2 DictRow and PL/Python rows support this.
"""
def keys(self) -> Iterable[str]: raise NotImplementedError
def values(self) -> Iterable[Any]: raise NotImplementedError
def items(self) -> Iterable[Tuple[str, Any]]: raise NotImplementedError
def get(self, key: str, default: Any = None) -> Any: raise NotImplementedError
def __getitem__(self, key: Union[str, int]) -> Any: raise NotImplementedError
def __iter__(self) -> Iterable[str]: raise NotImplementedError
def __len__(self) -> int: raise NotImplementedError
def __contains__(self, key: str) -> bool: raise NotImplementedError
class Cursor(Protocol):
@property
def rowcount(self) -> int: raise NotImplementedError
def execute(self, sql: str, params: Optional[ExecuteParams] = None) -> None: raise NotImplementedError
def fetchall(self) -> Sequence[DictRow]: raise NotImplementedError
def fetchone(self) -> DictRow: raise NotImplementedError
def __enter__(self) -> "Cursor": raise NotImplementedError
def __exit__(self, typ: Type, value: Any, traceback: Any) -> Any: raise NotImplementedError
def copy_expert(
self, sql: str,
f: Union[IO[str], IO[bytes], io.TextIOBase, io.RawIOBase],
size: int = 8192
) -> None:
raise NotImplementedError
def fileno(self) -> int: raise NotImplementedError
@property
def description(self) -> Sequence[Tuple[str, int, int, int, Optional[int], Optional[int], None]]: raise NotImplementedError
class Connection(Protocol):
def cursor(self) -> Cursor: raise NotImplementedError
def rollback(self) -> None: raise NotImplementedError
def commit(self) -> None: raise NotImplementedError
def close(self) -> None: raise NotImplementedError
@property
def isolation_level(self) -> int: raise NotImplementedError
def set_isolation_level(self, level: int) -> None: raise NotImplementedError
class Runnable(Protocol):
def run(self) -> None: raise NotImplementedError
class HasFileno(Protocol):
def fileno(self) -> int: raise NotImplementedError
FileDescriptor = int
FileDescriptorLike = Union[int, HasFileno]
|
