from collections.abc import Iterable from folkugat_web.dal.sql import Connection, get_connection from folkugat_web.model import search as search_model from folkugat_web.model import temes as model from . import conversion _tema_id_to_ngrams_cache: dict[int, search_model.NGrams] | None = None def get_tema_by_id(tema_id: int, con: Connection | None = None) -> model.Tema | None: query = """ SELECT id, title, alternatives, creation_date, modification_date, hidden FROM temes WHERE id = :id """ data = dict(id=tema_id) with get_connection(con) as con: cur = con.cursor() _ = cur.execute(query, data) row: conversion.TemaRowTuple = cur.fetchone() return conversion.row_to_tema(row) if row else None def get_temes_by_ids(tema_ids: list[int], con: Connection | None = None) -> list[model.Tema]: placeholders = ", ".join(["?" for _ in tema_ids]) query = f""" SELECT id, title, alternatives, creation_date, modification_date, hidden FROM temes WHERE id IN ({placeholders}) """ with get_connection(con) as con: cur = con.cursor() _ = cur.execute(query, tema_ids) rows: Iterable[conversion.TemaRowTuple] = cur.fetchall() return list(map(conversion.row_to_tema, rows)) def evict_tema_id_to_ngrams_cache(): global _tema_id_to_ngrams_cache _tema_id_to_ngrams_cache = None def get_tema_id_to_ngrams(con: Connection | None = None) -> dict[int, search_model.NGrams]: global _tema_id_to_ngrams_cache if _tema_id_to_ngrams_cache is None: _tema_id_to_ngrams_cache = _get_tema_id_to_ngrams(con) return _tema_id_to_ngrams_cache def _get_tema_id_to_ngrams(con: Connection | None = None) -> dict[int, search_model.NGrams]: query = """ SELECT id, ngrams FROM temes """ with get_connection(con) as con: cur = con.cursor() _ = cur.execute(query) rows: list[tuple[int, str]] = cur.fetchall() return {id_: conversion.cell_to_ngrams(ng) for id_, ng in rows}