49 lines
1.2 KiB
Python
49 lines
1.2 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 lyrics as lyrics_dal
|
|
from folkugat_web.model.temes import Lyrics
|
|
|
|
query = """ SELECT id, lyrics FROM temes """
|
|
LyricRowTuple = tuple[int, str]
|
|
|
|
|
|
class OldLyricRowDict(TypedDict):
|
|
title: str
|
|
content: str
|
|
|
|
|
|
def _build_lyric(tema_id: int, lyric_dict: OldLyricRowDict) -> Lyrics:
|
|
return Lyrics(
|
|
id=None,
|
|
tema_id=tema_id,
|
|
title=lyric_dict["title"],
|
|
content=lyric_dict["content"],
|
|
)
|
|
|
|
|
|
def _build_lyrics(row: LyricRowTuple) -> list[Lyrics]:
|
|
tema_id = row[0]
|
|
lyric_dicts: list[OldLyricRowDict] = json.loads(row[1])
|
|
return [_build_lyric(tema_id, lyric_dict) for lyric_dict in lyric_dicts]
|
|
|
|
|
|
with get_connection() as con:
|
|
cur = con.cursor()
|
|
_ = cur.execute(query)
|
|
iter_rows: Iterable[LyricRowTuple] = cur.fetchall()
|
|
rows = list(map(_build_lyrics, iter_rows))
|
|
|
|
_ = cur.execute("DELETE FROM tema_lyrics")
|
|
for lyrics in rows:
|
|
for lyric in lyrics:
|
|
_ = lyrics_dal.insert_lyric(lyric=lyric, con=con)
|
|
|
|
drop_query = """ ALTER TABLE temes DROP COLUMN lyrics"""
|
|
_ = cur.execute(drop_query)
|
|
|
|
|
|
print("DONE!")
|