Added lilypond edition

folkugat_web/dal/sql/temes/ddl.py

@@ -7,6 +7,7 @@ def create_db(con: Connection | None = None):
         create_links_table(con)
         create_lyrics_table(con)
         create_properties_table(con)
+        create_scores_table(con)
 
 
 def create_temes_table(con: Connection):
@@ -66,3 +67,21 @@ def create_properties_table(con: Connection):
     """
     cur = con.cursor()
     _ = cur.execute(query)
+
+
+def create_scores_table(con: Connection):
+    query = """
+      CREATE TABLE IF NOT EXISTS tema_scores (
+        id INTEGER PRIMARY KEY,
+        tema_id INTEGER,
+        title TEXT,
+        source TEXT NOT NULL,
+        errors TEXT NOT NULL,
+        img_url TEXT,
+        pdf_url TEXT,
+        hidden BOOLEAN,
+        FOREIGN KEY(tema_id) REFERENCES temes(id) ON DELETE CASCADE
+    )
+    """
+    cur = con.cursor()
+    _ = cur.execute(query)

folkugat_web/dal/sql/temes/scores.py

@@ -0,0 +1,133 @@
+import json
+from collections.abc import Iterable
+from typing import TypedDict
+
+from folkugat_web.dal.sql import Connection, get_connection
+from folkugat_web.model import lilypond as lilypond_model
+from folkugat_web.model import temes as model
+
+ScoreRowTuple = tuple[int, int, str, str, str, str | None, str | None, bool]
+
+
+class ScoreRowDict(TypedDict):
+    id: int | None
+    tema_id: int
+    title: str
+    errors: str
+    source: str
+    img_url: str | None
+    pdf_url: str | None
+    hidden: bool
+
+
+def score_to_row(score: model.Score) -> ScoreRowDict:
+    return {
+        "id": score.id,
+        "tema_id": score.tema_id,
+        "title": score.title,
+        "errors": json.dumps(list(error.to_dict() for error in score.errors)),
+        "source": score.source,
+        "img_url": score.img_url,
+        "pdf_url": score.pdf_url,
+        "hidden": score.hidden,
+    }
+
+
+def row_to_score(row: ScoreRowTuple) -> model.Score:
+    errors_dicts: list[dict[str, str]] = json.loads(row[4])
+    return model.Score(
+        id=row[0],
+        tema_id=row[1],
+        title=row[2],
+        source=row[3],
+        errors=list(map(lilypond_model.RenderError.from_dict, errors_dicts)),
+        img_url=row[5],
+        pdf_url=row[6],
+        hidden=bool(row[7]),
+    )
+
+
+class QueryData(TypedDict, total=False):
+    id: int
+    tema_id: int
+
+
+def _filter_clause(
+    score_id: int | None,
+    tema_id: int | None,
+) -> tuple[str, QueryData]:
+    filter_clauses: list[str] = []
+    filter_data: QueryData = {}
+
+    if score_id is not None:
+        filter_clauses.append("id = :id")
+        filter_data["id"] = score_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_scores(score_id: int | None = None, tema_id: int | None = None, con: Connection | None = None) -> Iterable[model.Score]:
+    filter_clause, data = _filter_clause(score_id=score_id, tema_id=tema_id)
+    if filter_clause:
+        filter_clause = f"WHERE {filter_clause}"
+
+    query = f"""
+      SELECT
+        id, tema_id, title, source, errors, img_url, pdf_url, hidden
+      FROM tema_scores
+      {filter_clause}
+    """
+    with get_connection(con) as con:
+        cur = con.cursor()
+        _ = cur.execute(query, data)
+        return map(row_to_score, cur.fetchall())
+
+
+def insert_score(score: model.Score, con: Connection | None = None) -> model.Score:
+    data = score_to_row(score)
+    query = f"""
+      INSERT INTO tema_scores
+        (id, tema_id, title, source, errors, img_url, pdf_url, hidden)
+      VALUES
+        (:id, :tema_id, :title, :source, :errors, :img_url, :pdf_url, :hidden)
+      RETURNING *
+    """
+    with get_connection(con) as con:
+        cur = con.cursor()
+        _ = cur.execute(query, data)
+        row: ScoreRowTuple = cur.fetchone()
+        return row_to_score(row)
+
+
+def update_score(score: model.Score, con: Connection | None = None):
+    data = score_to_row(score)
+    query = """
+      UPDATE tema_scores
+      SET
+        tema_id = :tema_id, title = :title, source = :source, errors = :errors,
+        img_url = :img_url, pdf_url = :pdf_url, hidden = :hidden
+      WHERE
+        id = :id
+    """
+    with get_connection(con) as con:
+        cur = con.cursor()
+        _ = cur.execute(query, data)
+
+
+def delete_score(score_id: int, tema_id: int | None = None, con: Connection | None = None):
+    filter_clause, data = _filter_clause(score_id=score_id, tema_id=tema_id)
+    if filter_clause:
+        filter_clause = f"WHERE {filter_clause}"
+
+    query = f"""
+      DELETE FROM tema_scores
+      {filter_clause}
+    """
+    with get_connection(con) as con:
+        cur = con.cursor()
+        _ = cur.execute(query, data)