138 lines
3.6 KiB
Python
138 lines
3.6 KiB
Python
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.utils import map_none
|
|
|
|
LinkRowTuple = tuple[int, int, str, str | None, str, str]
|
|
|
|
|
|
class LinkRowDict(TypedDict):
|
|
id: int | None
|
|
tema_id: int
|
|
content_type: str
|
|
link_type: str | None
|
|
title: str
|
|
url: str
|
|
|
|
|
|
def link_to_row(link: model.Link) -> LinkRowDict:
|
|
return {
|
|
"id": link.id,
|
|
"tema_id": link.tema_id,
|
|
"content_type": link.content_type.value,
|
|
"link_type": map_none(lambda lt: lt.value, link.link_type),
|
|
"title": link.title,
|
|
"url": link.url,
|
|
}
|
|
|
|
|
|
def row_to_link(row: LinkRowTuple) -> model.Link:
|
|
return model.Link(
|
|
id=row[0],
|
|
tema_id=row[1],
|
|
content_type=model.ContentType(row[2]),
|
|
link_type=map_none(model.LinkType, row[3]),
|
|
title=row[4],
|
|
url=row[5],
|
|
)
|
|
|
|
|
|
class QueryData(TypedDict, total=False):
|
|
id: int
|
|
tema_id: int
|
|
|
|
|
|
def _filter_clause(
|
|
link_id: int | None,
|
|
tema_id: int | None,
|
|
) -> tuple[str, QueryData]:
|
|
filter_clauses: list[str] = []
|
|
filter_data: QueryData = {}
|
|
|
|
if link_id is not None:
|
|
filter_clauses.append("id = :id")
|
|
filter_data["id"] = link_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_links(link_id: int | None = None, tema_id: int | None = None, con: Connection | None = None) -> Iterable[model.Link]:
|
|
filter_clause, data = _filter_clause(link_id=link_id, tema_id=tema_id)
|
|
if filter_clause:
|
|
filter_clause = f"WHERE {filter_clause}"
|
|
|
|
query = f"""
|
|
SELECT
|
|
id, tema_id, content_type, link_type, title, url
|
|
FROM tema_links
|
|
{filter_clause}
|
|
"""
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
return map(row_to_link, cur.fetchall())
|
|
|
|
|
|
def insert_link(link: model.Link, con: Connection | None = None) -> model.Link:
|
|
data = link_to_row(link)
|
|
query = f"""
|
|
INSERT INTO tema_links
|
|
(id, tema_id, content_type, link_type, title, url)
|
|
VALUES
|
|
(:id, :tema_id, :content_type, :link_type, :title, :url)
|
|
RETURNING *
|
|
"""
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
row: LinkRowTuple = cur.fetchone()
|
|
return row_to_link(row)
|
|
|
|
|
|
def create_link(tema_id: int, con: Connection | None = None) -> model.Link:
|
|
new_link = model.Link(
|
|
id=None,
|
|
tema_id=tema_id,
|
|
content_type=model.ContentType.PARTITURA,
|
|
link_type=None,
|
|
url="",
|
|
title="",
|
|
)
|
|
return insert_link(new_link, con=con)
|
|
|
|
|
|
def update_link(link: model.Link, con: Connection | None = None):
|
|
data = link_to_row(link)
|
|
query = """
|
|
UPDATE tema_links
|
|
SET
|
|
tema_id = :tema_id, content_type = :content_type, link_type = :link_type,
|
|
title = :title, url = :url
|
|
WHERE
|
|
id = :id
|
|
"""
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|
|
|
|
|
|
def delete_link(link_id: int, tema_id: int | None = None, con: Connection | None = None):
|
|
filter_clause, data = _filter_clause(link_id=link_id, tema_id=tema_id)
|
|
if filter_clause:
|
|
filter_clause = f"WHERE {filter_clause}"
|
|
|
|
query = f"""
|
|
DELETE FROM tema_links
|
|
{filter_clause}
|
|
"""
|
|
with get_connection(con) as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query, data)
|