36 lines
1.6 KiB
Python
36 lines
1.6 KiB
Python
from typing import Dict, List, Any
|
||
import aiosqlite
|
||
from app.config import settings
|
||
|
||
|
||
class SQLiteExecutor:
|
||
"""Executor для SQL запросов в БД SQLlite"""
|
||
def __init__(self) -> None:
|
||
self._db_path = settings.SQLITE_DATABASE_PATH
|
||
|
||
async def execute_query(
|
||
self, query: str, params: tuple = ()
|
||
) -> List[Dict[str, Any]]:
|
||
"""Выполнить запрос, который не должен вносить изменения в БД.
|
||
|
||
:param query: SQL запрос
|
||
:param params: Параметры, которые нужно передать вместе с запросом
|
||
:return: Список, выдаваемый aiosqlite
|
||
"""
|
||
async with aiosqlite.connect(self._db_path) as db:
|
||
db.row_factory = aiosqlite.Row
|
||
cursor = await db.execute(query, params)
|
||
rows = await cursor.fetchall()
|
||
return [dict(row) for row in rows]
|
||
|
||
async def execute_mod_query(self, query: str, params: tuple = ()) -> int:
|
||
"""Выполнить запрос, который что-то изменяет в БД.
|
||
|
||
:param query: SQL запрос
|
||
:param params: Параметры, которые нужно передать вместе с запросом
|
||
:return: Список, выдаваемый aiosqlite
|
||
"""
|
||
async with aiosqlite.connect(self._db_path) as db:
|
||
cursor = await db.execute(query, params)
|
||
await db.commit()
|
||
return cursor.lastrowid if cursor.lastrowid else cursor.rowcount
|