OpenBookr/openbookr/db.py

# -*- coding: utf-8 -*-

##################
# DataBase logic #
##################

from openbookr.ABC import IDataBase
from openbookr.models import Book
from aiosqlite import connect as sqdb, Connection as DB, Cursor
from enum import Enum
from typing import Any, Iterable


class DBColumnsTypes(Enum):
    """
    Data types for sqlite table columns
    """

    stroke = "TEXT"
    integer_number = "INTEGER"
    float_number = "FLOAT"


class SQLiteEngine(object):

    db: DB
    cursor: Cursor

    @staticmethod
    def __getcolumns(columns: dict[str, DBColumnsTypes]) -> str:
        """
        Converts dictionary to a string for a sqlite query

        :param columns: Dict["column_title": ColumnType]
        :return: Part of sqlite query
        """
        keys = tuple(columns.keys())
        values = tuple(columns.values())
        lst = zip(keys, values)
        lst = [
            f"{column_title.lower()} {column_type.value}"
            for column_title, column_type in lst
        ]
        return ", ".join(lst)

    def __init__(self, db_path: str) -> None:
        self.path = db_path

    async def __aenter__(self) -> "SQLiteEngine":
        self.db = sqdb(self.path)
        self.cursor = await self.db.cursor()
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:
        await self.db.commit()
        await self.db.close()

    async def _execute(self, request: str, params: Iterable | None = None) -> None:
        """
        Executes sqlite query

        :param request: Query text
        :param params: Additional parameters for the request
        """
        if params is None:
            await self.cursor.execute(request)
            return
        await self.cursor.execute(request, params)

    async def _create_table(
        self, table: str, columns: dict[str, DBColumnsTypes]
    ) -> None:
        """
        Generates table in the database

        :param table: Name of the table
        :param columns: Columns in the table
        """
        request = f"CREATE TABLE IF NOT EXISTS {table} (id INTEGER PRIMARY KEY, {self.__getcolumns(columns)})"
        await self._execute(request)

    async def _insert(self, table: str, items: dict[str, Any]) -> None:
        """
        Inserts data into a field of a database table

        :param table: Name of the table
        :param items: Dict["column_title": value]
        """
        columns = [title.lower() for title in tuple(items.keys())]
        columns = ", ".join(columns)
        values = list(items.values())
        request = (
            f"INSERT INTO {table} ({columns}) VALUES ({("?, " * len(values))[:-2]})"
        )
        await self._execute(request, values)

    async def _get(
        self,
        table: str,
        column: Iterable[str] | None = None,
        where: dict | str | None = None,
        order: str | None = None,
    ) -> Iterable[Any]:
        """
        Returns data from sqlite table

        :param table: Name of the table
        :param column: List of the columns titles
        :param where: Dict['column_title': value]
        :param order: sqlite query order syntax stroke
        :return: List of values from table
        """
        args: list[Any] | None = None
        if column is None:
            column = "*"
        else:
            column = ", ".join(i.lower() for i in column)
        request: list[Any] = [f"SELECT {column} FROM {table}"]
        if isinstance(where, dict):
            args = list(where.values())
            where = (
                f"WHERE {", ".join(tuple(where.keys()))} = ({("?, " * len(args))[:-2]})"
            )
            request.append(where)
        if order is not None:
            order = f"ORDER BY {order}"
            request.append(order)
        request = [" ".join(request)]
        if isinstance(args, list):
            request.append(args)
        await self.cursor.execute(*request)
        return await self.cursor.fetchall()

    async def _delete(self, table: str, where: dict[str, Any]) -> None:
        values = list(where.values())
        request = f"DELETE FROM {table} WHERE {", ".join(tuple(where.keys()))} = ({("?, " * len(values))[:-2]})"
        await self._execute(request, values)


class SQlite(IDataBase, SQLiteEngine):

    async def authors(self) -> Iterable:
        ...

    async def genres(self) -> Iterable:
        ...

    async def tags(self) -> Iterable:
        ...

    async def add_book(self, book: Book) -> None:
        ...

    async def get_book(self, book_id: int) -> Book:
        ...

    async def del_book(self, book_id: int) -> None:
        ...

    async def update_book(self, book_id: int, updated_book: Book) -> None:
        ...

    async def book_list(
        self,
        author: str | None = None,
        tags: Iterable | None = None,
        genres: Iterable | None = None,
    ) -> Iterable:
        ...

    async def add_genre(self, genre: str) -> None:
        ...

    async def add_tag(self, tag: str) -> None:
        ...

    async def del_genre(self, genre_id: int) -> None:
        ...

    async def del_tag(self, tag_id: int) -> None:
        ...

    async def add_author(self, author: str) -> None:
        ...

    async def del_author(self, author_id: int) -> None:
        ...