137 lines
3.9 KiB
Python
137 lines
3.9 KiB
Python
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 temes as model
|
|
from folkugat_web.model.lilypond.processing import RenderError
|
|
|
|
ScoreRowTuple = tuple[int, int, str, str, str, str | None, 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
|
|
preview_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,
|
|
"preview_url": score.preview_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(RenderError.from_dict, errors_dicts)),
|
|
img_url=row[5],
|
|
pdf_url=row[6],
|
|
preview_url=row[7],
|
|
hidden=bool(row[8]),
|
|
)
|
|
|
|
|
|
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, preview_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, preview_url, hidden)
|
|
VALUES
|
|
(:id, :tema_id, :title, :source, :errors, :img_url, :pdf_url, :preview_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, preview_url = :preview_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)
|