33 lines
841 B
Python
33 lines
841 B
Python
from typing import Any
|
|
from collections.abc import Sequence, Iterable
|
|
|
|
import aiosqlite
|
|
|
|
from ter.config import Settings
|
|
|
|
settings = Settings()
|
|
|
|
|
|
class Database:
|
|
Params = Sequence[Any] | dict[str, Any]
|
|
|
|
def __init__(self, uri: str) -> None:
|
|
self._uri: str = uri
|
|
self._connection: aiosqlite.Connection | None = None
|
|
|
|
async def connect(self):
|
|
self._connection = await aiosqlite.connect(self._uri, uri=True)
|
|
|
|
async def disconnect(self):
|
|
await self._connection.close()
|
|
self._connection = None
|
|
|
|
async def execute(self, sql: str, params: Params = ()):
|
|
return await self._connection.execute(sql, params)
|
|
|
|
async def executemany(self, sql: str, params: Iterable[Params]):
|
|
return await self._connection.executemany(sql, params)
|
|
|
|
|
|
database = Database(settings.SQLITE_URI)
|