from collections.abc import Iterable from typing import TypedDict from folkugat_web.dal.sql import Connection, get_connection from folkugat_web.model import temes as model LyricRowTuple = tuple[int, int, str, str, int | None] class LyricRowDict(TypedDict): id: int | None tema_id: int title: str content: str max_columns: int | None def lyric_to_row(lyric: model.Lyrics) -> LyricRowDict: return { "id": lyric.id, "tema_id": lyric.tema_id, "title": lyric.title, "content": lyric.content, "max_columns": lyric.max_columns, } def row_to_lyric(row: LyricRowTuple) -> model.Lyrics: return model.Lyrics( id=row[0], tema_id=row[1], title=row[2], content=row[3], max_columns=row[4], ) class QueryData(TypedDict, total=False): id: int tema_id: int def _filter_clause( lyric_id: int | None, tema_id: int | None, ) -> tuple[str, QueryData]: filter_clauses: list[str] = [] filter_data: QueryData = {} if lyric_id is not None: filter_clauses.append("id = :id") filter_data["id"] = lyric_id if tema_id is not None: filter_clauses.append("tema_id = :tema_id") filter_data["tema_id"] = tema_id filter_clause = " AND ".join(filter_clauses) return filter_clause, filter_data def get_lyrics(lyric_id: int | None = None, tema_id: int | None = None, con: Connection | None = None) -> Iterable[model.Lyrics]: filter_clause, data = _filter_clause(lyric_id=lyric_id, tema_id=tema_id) if filter_clause: filter_clause = f"WHERE {filter_clause}" query = f""" SELECT id, tema_id, title, content, max_columns FROM tema_lyrics {filter_clause} """ with get_connection(con) as con: cur = con.cursor() _ = cur.execute(query, data) return map(row_to_lyric, cur.fetchall()) def insert_lyric(lyric: model.Lyrics, con: Connection | None = None) -> model.Lyrics: data = lyric_to_row(lyric) query = f""" INSERT INTO tema_lyrics (id, tema_id, title, content, max_columns) VALUES (:id, :tema_id, :title, :content, :max_columns) RETURNING * """ with get_connection(con) as con: cur = con.cursor() _ = cur.execute(query, data) row: LyricRowTuple = cur.fetchone() return row_to_lyric(row) def create_lyric(tema_id: int, title: str | None = None, con: Connection | None = None) -> model.Lyrics: new_lyric = model.Lyrics( id=None, tema_id=tema_id, title=title or "", content="", max_columns=None, ) return insert_lyric(new_lyric, con=con) def update_lyric(lyric: model.Lyrics, con: Connection | None = None): data = lyric_to_row(lyric) query = """ UPDATE tema_lyrics SET tema_id = :tema_id, title = :title, content = :content, max_columns = :max_columns WHERE id = :id """ with get_connection(con) as con: cur = con.cursor() _ = cur.execute(query, data) def delete_lyric(lyric_id: int, tema_id: int | None = None, con: Connection | None = None): filter_clause, data = _filter_clause(lyric_id=lyric_id, tema_id=tema_id) if filter_clause: filter_clause = f"WHERE {filter_clause}" query = f""" DELETE FROM tema_lyrics {filter_clause} """ with get_connection(con) as con: cur = con.cursor() _ = cur.execute(query, data)