140 lines
3.2 KiB
Python
140 lines
3.2 KiB
Python
from folkugat_web.dal.sql import Connection, get_connection
|
|
from folkugat_web.model import playlists as model
|
|
|
|
from . import conversion
|
|
|
|
|
|
def create_playlist(
|
|
name: str | None = None,
|
|
con: Connection | None = None,
|
|
) -> int:
|
|
query = """
|
|
INSERT INTO playlists
|
|
(name)
|
|
VALUES
|
|
(:name)
|
|
RETURNING *
|
|
"""
|
|
data = dict(name=name)
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
row: conversion.PlaylistRowTuple = cur.fetchone()
|
|
return row[0]
|
|
|
|
|
|
def delete_playlist(
|
|
playlist_id: int,
|
|
con: Connection | None = None,
|
|
) -> None:
|
|
query = """
|
|
DELETE FROM playlists
|
|
WHERE id = :id
|
|
"""
|
|
data = dict(id=playlist_id)
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
|
|
|
|
def insert_playlist_entry(
|
|
pl_entry: model.PlaylistEntry,
|
|
con: Connection | None = None,
|
|
) -> model.PlaylistEntry:
|
|
query = """
|
|
INSERT INTO playlist_entries
|
|
(id, playlist_id, set_id, tema_id)
|
|
VALUES
|
|
(:id, :playlist_id, :set_id, :tema_id)
|
|
RETURNING *
|
|
"""
|
|
data = conversion.playlist_entry_to_row(pl_entry)
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
row: conversion.PlaylistEntryRowTuple = cur.fetchone()
|
|
return conversion.row_to_playlist_entry(row)
|
|
|
|
|
|
def update_playlist_entry(
|
|
entry: model.PlaylistEntry,
|
|
con: Connection | None = None,
|
|
):
|
|
query = """
|
|
UPDATE playlist_entries
|
|
SET
|
|
id = :id, playlist_id = :playlist_id, set_id = :set_id, tema_id = :tema_id
|
|
WHERE
|
|
id = :id
|
|
"""
|
|
data = conversion.playlist_entry_to_row(entry)
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
return
|
|
|
|
|
|
def delete_playlist_entry(
|
|
entry_id: int,
|
|
con: Connection | None = None,
|
|
):
|
|
query = """
|
|
DELETE FROM playlist_entries
|
|
WHERE id = :id
|
|
"""
|
|
data = dict(id=entry_id)
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
return
|
|
|
|
|
|
def delete_playlist_set(
|
|
playlist_id: int,
|
|
set_id: int,
|
|
con: Connection | None = None,
|
|
):
|
|
query = """
|
|
DELETE FROM playlist_entries
|
|
WHERE playlist_id = :playlist_id AND set_id = :set_id
|
|
"""
|
|
data = dict(playlist_id=playlist_id, set_id=set_id)
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
return
|
|
|
|
|
|
def update_playlist_name(
|
|
playlist_id: int,
|
|
name: str | None,
|
|
con: Connection | None = None,
|
|
):
|
|
query = """
|
|
UPDATE playlists SET
|
|
name = :name
|
|
WHERE id = :id
|
|
"""
|
|
data = dict(id=playlist_id, name=name)
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
return
|
|
|
|
|
|
def update_playlist_visibility(
|
|
playlist_id: int,
|
|
hidden: bool,
|
|
con: Connection | None = None,
|
|
):
|
|
query = """
|
|
UPDATE playlists SET
|
|
hidden = :hidden
|
|
WHERE id = :id
|
|
"""
|
|
data = dict(id=playlist_id, hidden=1 if hidden else 0)
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
return
|