54 lines
1.4 KiB
Python
54 lines
1.4 KiB
Python
import json
|
|
from collections.abc import Iterable
|
|
from typing import TypedDict
|
|
|
|
from folkugat_web.dal.sql import get_connection
|
|
from folkugat_web.dal.sql.temes import links as links_dal
|
|
from folkugat_web.model.temes import ContentType, Link, LinkType
|
|
from folkugat_web.utils import map_none
|
|
|
|
query = """ SELECT id, links FROM temes """
|
|
LinkRowTuple = tuple[int, str]
|
|
|
|
|
|
class OldLinkRowDict(TypedDict):
|
|
content_type: str
|
|
link_type: str | None
|
|
title: str
|
|
url: str
|
|
|
|
|
|
def _build_link(tema_id: int, link_dict: OldLinkRowDict) -> Link:
|
|
return Link(
|
|
id=None,
|
|
tema_id=tema_id,
|
|
content_type=ContentType(link_dict["content_type"]),
|
|
link_type=map_none(LinkType, link_dict.get("link_type")),
|
|
url=link_dict["url"],
|
|
title=link_dict["title"],
|
|
)
|
|
|
|
|
|
def _build_links(row: LinkRowTuple) -> list[Link]:
|
|
tema_id = row[0]
|
|
link_dicts: list[OldLinkRowDict] = json.loads(row[1])
|
|
return [_build_link(tema_id, link_dict) for link_dict in link_dicts]
|
|
|
|
|
|
with get_connection() as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query)
|
|
iter_rows: Iterable[LinkRowTuple] = cur.fetchall()
|
|
rows = list(map(_build_links, iter_rows))
|
|
|
|
_ = cur.execute("DELETE FROM tema_links")
|
|
for links in rows:
|
|
for link in links:
|
|
_ = links_dal.insert_link(link=link, con=con)
|
|
|
|
drop_query = """ ALTER TABLE temes DROP COLUMN links"""
|
|
_ = cur.execute(drop_query)
|
|
|
|
|
|
print("DONE!")
|