This repository was archived by the owner on Aug 19, 2025. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 268
Expand file tree
/
Copy pathinterfaces.py
More file actions
78 lines (55 loc) 路 2.51 KB
/
Copy pathinterfaces.py
File metadata and controls
78 lines (55 loc) 路 2.51 KB
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
73
74
75
76
77
78
import typing
from collections.abc import Sequence
from sqlalchemy.sql import ClauseElement
class DatabaseBackend:
async def connect(self) -> None:
raise NotImplementedError() # pragma: no cover
async def disconnect(self) -> None:
raise NotImplementedError() # pragma: no cover
def connection(self) -> "ConnectionBackend":
raise NotImplementedError() # pragma: no cover
class ConnectionBackend:
async def acquire(self) -> None:
raise NotImplementedError() # pragma: no cover
async def release(self) -> None:
raise NotImplementedError() # pragma: no cover
async def fetch_all(self, query: ClauseElement) -> typing.List["Record"]:
raise NotImplementedError() # pragma: no cover
async def fetch_one(self, query: ClauseElement) -> typing.Optional["Record"]:
raise NotImplementedError() # pragma: no cover
async def fetch_val(
self, query: ClauseElement, column: typing.Any = 0
) -> typing.Any:
row = await self.fetch_one(query)
return None if row is None else row[column]
async def execute(self, query: ClauseElement) -> typing.Any:
raise NotImplementedError() # pragma: no cover
async def execute_many(self, queries: typing.List[ClauseElement]) -> None:
raise NotImplementedError() # pragma: no cover
async def iterate(
self, query: ClauseElement
) -> typing.AsyncGenerator[typing.Mapping, None]:
raise NotImplementedError() # pragma: no cover
# mypy needs async iterators to contain a `yield`
# https://github.com/python/mypy/issues/5385#issuecomment-407281656
yield True # pragma: no cover
def transaction(self) -> "TransactionBackend":
raise NotImplementedError() # pragma: no cover
@property
def raw_connection(self) -> typing.Any:
raise NotImplementedError() # pragma: no cover
class TransactionBackend:
async def start(
self, is_root: bool, extra_options: typing.Dict[typing.Any, typing.Any]
) -> None:
raise NotImplementedError() # pragma: no cover
async def commit(self) -> None:
raise NotImplementedError() # pragma: no cover
async def rollback(self) -> None:
raise NotImplementedError() # pragma: no cover
class Record(Sequence):
@property
def _mapping(self) -> typing.Mapping:
raise NotImplementedError() # pragma: no cover
def __getitem__(self, key: typing.Any) -> typing.Any:
raise NotImplementedError() # pragma: no cover