Add SQLiteEngine class logic
This commit is contained in:
parent
272a574aef
commit
79dcf8c1aa
150
openbookr/db.py
150
openbookr/db.py
|
@ -1,5 +1,4 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
from typing import Iterable
|
||||
|
||||
##################
|
||||
# DataBase logic #
|
||||
|
@ -7,31 +6,166 @@ from typing import Iterable
|
|||
|
||||
from openbookr.ABC import IDataBase
|
||||
from openbookr.models import Book
|
||||
from aiosqlite import connect as sqdb
|
||||
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):
|
||||
...
|
||||
|
||||
@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":
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc, tb) -> None:
|
||||
...
|
||||
|
||||
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
|
||||
"""
|
||||
async with sqdb(self.path) as db:
|
||||
cursor = await db.cursor()
|
||||
if params is None:
|
||||
await cursor.execute(request)
|
||||
return
|
||||
await cursor.execute(request, params)
|
||||
await db.commit()
|
||||
await db.close()
|
||||
|
||||
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)
|
||||
async with sqdb(self.path) as db:
|
||||
cursor = await db.cursor()
|
||||
await cursor.execute(*request)
|
||||
return await 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):
|
||||
class SQlite(IDataBase):
|
||||
|
||||
def __init__(self, db_path: str) -> None:
|
||||
self.path = db_path
|
||||
|
||||
async def __getsmth(self, table: str) -> Iterable:
|
||||
async with SQLiteEngine(self.path) as db:
|
||||
return await db.get(table=table)
|
||||
|
||||
async def authors(self) -> Iterable:
|
||||
...
|
||||
return await self.__getsmth("authors")
|
||||
|
||||
async def genres(self) -> Iterable:
|
||||
...
|
||||
return await self.__getsmth("genres")
|
||||
|
||||
async def tags(self) -> Iterable:
|
||||
...
|
||||
return await self.__getsmth("tags")
|
||||
|
||||
async def add_book(self, book: Book) -> None:
|
||||
...
|
||||
obj = dict(book.model_dump())
|
||||
async with SQLiteEngine(self.path) as db:
|
||||
await db.insert(table="books", items=obj)
|
||||
|
||||
async def get_book(self, book_id: int) -> Book:
|
||||
...
|
||||
|
||||
async def del_book(self, book_id: int) -> None:
|
||||
...
|
||||
async with SQLiteEngine(self.path) as db:
|
||||
await db.delete(table="books", where={"id": book_id})
|
||||
|
||||
async def update_book(self, book_id: int, updated_book: Book) -> None:
|
||||
...
|
||||
|
|
|
@ -1,3 +1,4 @@
|
|||
aiosqlite==0.20.0
|
||||
annotated-types==0.7.0
|
||||
pydantic==2.8.2
|
||||
pydantic_core==2.20.1
|
||||
|
|
Loading…
Reference in New Issue