111 lines
3.2 KiB
Python
111 lines
3.2 KiB
Python
from collections.abc import Iterator
|
|
from typing import TypedDict
|
|
|
|
from folkugat_web.dal.sql import Connection, get_connection
|
|
from folkugat_web.dal.sql.sessions import conversion as sessions_conversion
|
|
from folkugat_web.model import playlists as model
|
|
from folkugat_web.model.sessions import Session
|
|
from folkugat_web.utils import groupby
|
|
|
|
from . import conversion
|
|
|
|
|
|
class QueryData(TypedDict, total=False):
|
|
id: int
|
|
set_id: int
|
|
playlist_id: int
|
|
|
|
|
|
def _filter_clause(
|
|
entry_id: int | None = None,
|
|
set_id: int | None = None,
|
|
playlist_id: int | None = None,
|
|
) -> tuple[str, QueryData]:
|
|
filter_clauses: list[str] = []
|
|
query_data: QueryData = {}
|
|
|
|
if entry_id is not None:
|
|
filter_clauses.append("id = :id")
|
|
query_data["id"] = entry_id
|
|
if set_id is not None:
|
|
filter_clauses.append("set_id = :set_id")
|
|
query_data["set_id"] = set_id
|
|
if playlist_id is not None:
|
|
filter_clauses.append("playlist_id = :playlist_id")
|
|
query_data["playlist_id"] = playlist_id
|
|
|
|
return " AND ".join(filter_clauses), query_data
|
|
|
|
|
|
def get_playlist_entries(
|
|
entry_id: int | None = None,
|
|
set_id: int | None = None,
|
|
playlist_id: int | None = None,
|
|
con: Connection | None = None,
|
|
) -> Iterator[model.PlaylistEntry]:
|
|
filter_clause, data = _filter_clause(entry_id=entry_id, set_id=set_id, playlist_id=playlist_id)
|
|
query = f"""
|
|
SELECT
|
|
id, playlist_id, set_id, tema_id
|
|
FROM playlist_entries
|
|
WHERE {filter_clause}
|
|
ORDER BY id ASC
|
|
"""
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
return map(conversion.row_to_playlist_entry, cur.fetchall())
|
|
|
|
|
|
def get_playlist_name(playlist_id: int, con: Connection | None = None) -> str | None:
|
|
query = """
|
|
SELECT name
|
|
FROM playlists
|
|
WHERE id = :playlist_id
|
|
"""
|
|
data = dict(playlist_id=playlist_id)
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
row = cur.fetchone()
|
|
return row[0] if row else None
|
|
|
|
|
|
def get_all_playlists(logged_in: bool = False, con: Connection | None = None) -> Iterator[model.Playlist]:
|
|
if logged_in:
|
|
# Show all playlists for logged in users, except session-associated ones
|
|
query = """
|
|
SELECT id, name, hidden
|
|
FROM playlists
|
|
WHERE id NOT IN (
|
|
SELECT DISTINCT playlist_id
|
|
FROM session_playlists
|
|
)
|
|
ORDER BY id DESC
|
|
"""
|
|
else:
|
|
# Show only visible playlists for non-logged in users, except session-associated ones
|
|
query = """
|
|
SELECT id, name, hidden
|
|
FROM playlists
|
|
WHERE hidden = 0 AND id NOT IN (
|
|
SELECT DISTINCT playlist_id
|
|
FROM session_playlists
|
|
)
|
|
ORDER BY id DESC
|
|
"""
|
|
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query)
|
|
rows = cur.fetchall()
|
|
|
|
for row in rows:
|
|
# Convert to Playlist model without sets for list view
|
|
yield model.Playlist(
|
|
id=row[0],
|
|
name=row[1],
|
|
sets=[], # Empty sets list for now
|
|
hidden=bool(row[2])
|
|
)
|