44 lines
1.0 KiB
Python
44 lines
1.0 KiB
Python
"""Abstract realiztion for DB"""
|
|
|
|
from collections.abc import Awaitable, Callable
|
|
from contextlib import AbstractAsyncContextManager as AsyncContextManager
|
|
from typing import Any
|
|
|
|
from server.config import AppConfig
|
|
|
|
type ExecuteFun = Callable[[Any], Awaitable[None]]
|
|
|
|
|
|
class ConnectError(Exception):
|
|
"""Custom error wor failed connections"""
|
|
|
|
|
|
class AbstractSession:
|
|
async def __aenter__(self) -> "AbstractSession":
|
|
raise NotImplementedError
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
raise NotImplementedError
|
|
|
|
def begin(self) -> AsyncContextManager:
|
|
raise NotImplementedError
|
|
|
|
async def execute(self, data: Any):
|
|
raise NotImplementedError
|
|
|
|
async def commit(self):
|
|
raise NotImplementedError
|
|
|
|
|
|
class AbstractDB:
|
|
"""Abstract realiztion for DB"""
|
|
|
|
def __init__(self, cnf: AppConfig):
|
|
raise NotImplementedError
|
|
|
|
def session_master(self) -> AbstractSession:
|
|
raise NotImplementedError
|
|
|
|
def session_slave(self) -> AbstractSession:
|
|
raise NotImplementedError
|