1import typing 2 3from sqlalchemy.sql import ClauseElement 4 5 6class DatabaseBackend: 7 async def connect(self) -> None: 8 raise NotImplementedError() # pragma: no cover 9 10 async def disconnect(self) -> None: 11 raise NotImplementedError() # pragma: no cover 12 13 def connection(self) -> "ConnectionBackend": 14 raise NotImplementedError() # pragma: no cover 15 16 17class ConnectionBackend: 18 async def acquire(self) -> None: 19 raise NotImplementedError() # pragma: no cover 20 21 async def release(self) -> None: 22 raise NotImplementedError() # pragma: no cover 23 24 async def fetch_all(self, query: ClauseElement) -> typing.List[typing.Mapping]: 25 raise NotImplementedError() # pragma: no cover 26 27 async def fetch_one(self, query: ClauseElement) -> typing.Optional[typing.Mapping]: 28 raise NotImplementedError() # pragma: no cover 29 30 async def fetch_val( 31 self, query: ClauseElement, column: typing.Any = 0 32 ) -> typing.Any: 33 row = await self.fetch_one(query) 34 return None if row is None else row[column] 35 36 async def execute(self, query: ClauseElement) -> typing.Any: 37 raise NotImplementedError() # pragma: no cover 38 39 async def execute_many(self, queries: typing.List[ClauseElement]) -> None: 40 raise NotImplementedError() # pragma: no cover 41 42 async def iterate( 43 self, query: ClauseElement 44 ) -> typing.AsyncGenerator[typing.Mapping, None]: 45 raise NotImplementedError() # pragma: no cover 46 # mypy needs async iterators to contain a `yield` 47 # https://github.com/python/mypy/issues/5385#issuecomment-407281656 48 yield True # pragma: no cover 49 50 def transaction(self) -> "TransactionBackend": 51 raise NotImplementedError() # pragma: no cover 52 53 @property 54 def raw_connection(self) -> typing.Any: 55 raise NotImplementedError() # pragma: no cover 56 57 58class TransactionBackend: 59 async def start( 60 self, is_root: bool, extra_options: typing.Dict[typing.Any, typing.Any] 61 ) -> None: 62 raise NotImplementedError() # pragma: no cover 63 64 async def commit(self) -> None: 65 raise NotImplementedError() # pragma: no cover 66 67 async def rollback(self) -> None: 68 raise NotImplementedError() # pragma: no cover 69