databases/databases/interfaces.py at drop-python3.6 · pythonran/databases · GitHub

History

75 lines (53 loc) · 2.39 KB

Raw

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

databases/databases/interfaces.py at drop-python3.6 · pythonran/databases · GitHub

Expand file tree

Breadcrumbs

interfaces.py

Latest commit

History

Breadcrumbs

interfaces.py

File metadata and controls