diff --git a/utils/db_api/__init__.py b/utils/db_api/__init__.py index e69de29..8b13789 100644 --- a/utils/db_api/__init__.py +++ b/utils/db_api/__init__.py @@ -0,0 +1 @@ + diff --git a/utils/db_api/storages/__init__.py b/utils/db_api/storages/__init__.py new file mode 100644 index 0000000..5b34153 --- /dev/null +++ b/utils/db_api/storages/__init__.py @@ -0,0 +1 @@ +from .sqlite import SqliteConnection diff --git a/utils/db_api/storages/basestorage/__init__.py b/utils/db_api/storages/basestorage/__init__.py new file mode 100644 index 0000000..f6b223c --- /dev/null +++ b/utils/db_api/storages/basestorage/__init__.py @@ -0,0 +1 @@ +from . import storage diff --git a/utils/db_api/storages/basestorage/storage.py b/utils/db_api/storages/basestorage/storage.py new file mode 100644 index 0000000..93ae807 --- /dev/null +++ b/utils/db_api/storages/basestorage/storage.py @@ -0,0 +1,41 @@ +from typing import List, Type, TypeVar, Union + +T = TypeVar("T") + + +class RawConnection: + @staticmethod + def __make_request( + sql: str, + params: Union[tuple, List[tuple]] = None, + fetch: bool = False, + mult: bool = False + ): + """ + You have to override this method for all synchronous databases (e.g., Sqlite). + :param sql: + :param params: + :param fetch: + :param mult: + :return: + """ + raise NotImplementedError + + @staticmethod + def _make_request( + sql: str, + params: Union[tuple, List[tuple]] = None, + fetch: bool = False, + mult: bool = False, + model_type: Type[T] = None + ): + """ + You have to override this method for all synchronous databases (e.g., Sqlite). + :param sql: + :param params: + :param fetch: + :param mult: + :param model_type: + :return: + """ + raise NotImplementedError diff --git a/utils/db_api/storages/sqlite/__init__.py b/utils/db_api/storages/sqlite/__init__.py new file mode 100644 index 0000000..502dfdb --- /dev/null +++ b/utils/db_api/storages/sqlite/__init__.py @@ -0,0 +1 @@ +from .storage import SqliteConnection diff --git a/utils/db_api/storages/sqlite/storage.py b/utils/db_api/storages/sqlite/storage.py new file mode 100644 index 0000000..c766c50 --- /dev/null +++ b/utils/db_api/storages/sqlite/storage.py @@ -0,0 +1,85 @@ +import sqlite3 +from pathlib import Path +from typing import Any, Dict, List, Optional, Type, TypeVar, Union + +from loguru import logger + +from ..basestorage.storage import RawConnection + +T = TypeVar("T") +db_title = Path(__file__).parent.parent.parent.parent.parent / 'data' / 'db.db' # db file supposed to be in data folder + + +class SqliteDBConn: + def __init__(self, db_name): + self.db_name = db_name + + def __enter__(self): + self.conn = sqlite3.connect(self.db_name) + self.conn.row_factory = sqlite3.Row + return self.conn + + def __exit__(self, exc_type, exc_val, exc_tb): + self.conn.close() + if exc_val: + raise + + +class SqliteConnection(RawConnection): + @staticmethod + def __make_request( + sql: str, + params: Union[tuple, List[tuple]] = None, + fetch: bool = False, + mult: bool = False + ) -> Optional[Union[List[Dict[str, Any]], Dict[str, Any]]]: + with SqliteDBConn(db_title) as conn: + c = conn.cursor() + try: + if isinstance(params, list): + c.executemany(sql, params) + else: + c.execute(sql, params) + except Exception as e: + logger.error(e) + if fetch: + if mult: + r = c.fetchall() + else: + r = c.fetchone() + return r + else: + conn.commit() + + @staticmethod + def _convert_to_model(data: Optional[dict], model: Type[T]) -> Optional[T]: + if data is not None: + return model(**data) + else: + return None + + @staticmethod + def _make_request( + sql: str, + params: Union[tuple, List[tuple]] = None, + fetch: bool = False, + mult: bool = False, + model_type: Type[T] = None + ) -> Optional[Union[List[T], T]]: + raw = SqliteConnection.__make_request(sql, params, fetch, mult) + if raw is None: + if mult: + return [] + else: + return None + else: + if mult: + if model_type is not None: + return [SqliteConnection._convert_to_model(i, model_type) for i in raw] + else: + return [i for i in raw] + else: + if model_type is not None: + return SqliteConnection._convert_to_model(raw, model_type) + else: + return raw