Desacoblar playlists de sessions i afegir slow jams

folkugat_web/dal/sql/sessions/playlists.py

@@ -0,0 +1,138 @@
+from folkugat_web.dal.sql import Connection, get_connection
+from folkugat_web.dal.sql.temes.conversion import row_to_tema
+from folkugat_web.model import sessions as sessions_model
+from folkugat_web.model import temes as temes_model
+from folkugat_web.model.playlists import PlaylistType
+from folkugat_web.utils import groupby
+
+from . import conversion
+
+
+def get_playlist_id(
+    session_id: int,
+    playlist_type: PlaylistType,
+    con: Connection | None = None,
+) -> int | None:
+    query = f"""
+      SELECT playlist_id
+      FROM session_playlists
+      WHERE
+        session_id = :session_id AND
+        playlist_type = :playlist_type
+    """
+    data = dict(
+        session_id=session_id,
+        playlist_type=playlist_type.value,
+    )
+    with get_connection(con) as con:
+        cur = con.cursor()
+        _ = cur.execute(query, data)
+        if row := cur.fetchone():
+            return row[0]
+        return None
+
+
+def insert_playlist(
+    session_id: int,
+    playlist_type: PlaylistType,
+    playlist_id: int,
+    con: Connection | None = None,
+):
+    query = f"""
+      INSERT INTO session_playlists
+        (session_id, playlist_type, playlist_id)
+      VALUES
+        (:session_id, :playlist_type, :playlist_id)
+    """
+    data = dict(
+        session_id=session_id,
+        playlist_type=playlist_type.value,
+        playlist_id=playlist_id,
+    )
+    with get_connection(con) as con:
+        cur = con.cursor()
+        _ = cur.execute(query, data)
+
+
+def delete_playlist(
+    session_id: int,
+    playlist_type: PlaylistType,
+    con: Connection | None = None,
+):
+    query = f"""
+      DELETE FROM session_playlists
+      WHERE
+        session_id = :session_id AND
+        playlist_type = :playlist_type
+    """
+    data = dict(
+        session_id=session_id,
+        playlist_type=playlist_type.value,
+    )
+    with get_connection(con) as con:
+        cur = con.cursor()
+        _ = cur.execute(query, data)
+
+
+GetTuneSessionsRow = tuple[int, int, str, str, str, str | None, str | None, str | None, str | None, bool]
+
+
+def get_tune_sessions(
+    tema_ids: list[int],
+    con: Connection | None = None,
+) -> dict[int, list[sessions_model.Session]]:
+    placeholders = ", ".join(["?" for _ in tema_ids])
+    query = f"""
+      SELECT
+        p.tema_id, s.id, s.date, s.start_time, s.end_time, s.venue_name,
+        s.venue_url, s.notes, s.cartell_url, s.is_live
+      FROM playlist_entries p
+        JOIN session_playlists sp ON p.playlist_id = sp.playlist_id
+        JOIN sessions s ON sp.session_id = s.id
+      WHERE
+        p.tema_id IN ({placeholders}) AND
+        sp.playlist_type = "{PlaylistType.SESSION_SETLIST.value}"
+    """
+    with get_connection(con) as con:
+        cur = con.cursor()
+        _ = cur.execute(query, tema_ids)
+        result_rows: list[GetTuneSessionsRow] = cur.fetchall()
+        return dict(groupby(
+            result_rows,
+            key_fn=lambda row: row[0],
+            group_fn=lambda rows: list(conversion.row_to_session(row[1:]) for row in rows)
+        ))
+
+
+CommonlyPlayedTuneRow = tuple[int, str, str, str, str, int, int]
+
+
+def get_commonly_played_tunes(
+    tema_id: int,
+    con: Connection | None = None,
+) -> list[temes_model.CommonlyPlayedTema]:
+    query = f"""
+        SELECT
+            id, title, alternatives, creation_date, modification_date, hidden, count
+        FROM (
+            SELECT tema_id, count(*) count FROM playlist_entries p JOIN (
+                SELECT pe.playlist_id, pe.set_id
+                FROM playlist_entries pe JOIN session_playlists sp USING (playlist_id)
+                WHERE tema_id = :tema_id AND playlist_type = :playlist_type
+            ) s
+            ON p.playlist_id == s.playlist_id AND p.set_id == s.set_id
+            WHERE tema_id != :tema_id
+            GROUP BY tema_id
+        ) common JOIN temes t ON common.tema_id == t.id
+    """
+    data = dict(tema_id=tema_id, playlist_type=PlaylistType.SESSION_SETLIST.value)
+    with get_connection(con) as con:
+        cur = con.cursor()
+        _ = cur.execute(query, data)
+        result_rows: list[CommonlyPlayedTuneRow] = cur.fetchall()
+        return [
+            temes_model.CommonlyPlayedTema(
+                tema=row_to_tema(row[:6]),
+                count=row[6],
+            ) for row in result_rows
+        ]