database/engine.py

"""SQLite database engine with WAL mode and async support."""

import aiosqlite
from pathlib import Path
from typing import Optional

from config import config


_DB: Optional[aiosqlite.Connection] = None


async def get_db() -> aiosqlite.Connection:
    global _DB
    if _DB is None:
        _DB = await aiosqlite.connect(config.database.path)
        _DB.row_factory = aiosqlite.Row
        if config.database.wal_mode:
            await _DB.execute("PRAGMA journal_mode=WAL")
        await _DB.execute(f"PRAGMA busy_timeout={config.database.busy_timeout}")
        await _DB.execute("PRAGMA foreign_keys=ON")
    return _DB


async def close_db():
    global _DB
    if _DB is not None:
        await _DB.close()
        _DB = None


async def init_db():
    from database.migrations import run_migrations
    db = await get_db()
    await run_migrations(db)


async def execute(query: str, params: tuple = ()) -> aiosqlite.Cursor:
    db = await get_db()
    cursor = await db.execute(query, params)
    await db.commit()
    return cursor


async def fetch_one(query: str, params: tuple = ()) -> Optional[dict]:
    db = await get_db()
    cursor = await db.execute(query, params)
    row = await cursor.fetchone()
    return dict(row) if row else None


async def fetch_all(query: str, params: tuple = ()) -> list[dict]:
    db = await get_db()
    cursor = await db.execute(query, params)
    rows = await cursor.fetchall()
    return [dict(r) for r in rows]


async def execute_many(query: str, params_list: list[tuple]) -> None:
    db = await get_db()
    await db.executemany(query, params_list)
    await db.commit()