from typing import List, Optional from db import db from external_api.base.search_models import PlaylistImageModel from models import Image, Playlist from playlists.schemas import PlaylistResponseSchema class PlaylistsRepository: def __get_playlist(self, id: str) -> Optional[Playlist]: return Playlist.query.filter(Playlist.external_id == id).one_or_none() def __map_image(self, image: PlaylistImageModel) -> Image: return Image(url=image.url, width=image.width, height=image.height) def __create_playlist(self, playlist: PlaylistResponseSchema) -> Playlist: playlist_model = Playlist() playlist_model.name = playlist.name playlist_model.external_id = playlist.id playlist_model.owner_name = playlist.ownerName playlist_model.provider = playlist.provider playlist_model.images = [self.__map_image(image) for image in playlist.images] return playlist_model def __update_playlist(self, playlist: Playlist, updated_playlist: PlaylistResponseSchema) -> Playlist: if playlist.images: [db.session.delete(image) for image in playlist.images] playlist.images = [self.__map_image(image) for image in updated_playlist.images] playlist.owner_name = updated_playlist.ownerName playlist.provider = updated_playlist.provider playlist.name = updated_playlist.name return playlist def get_or_create_playlists(self, playlists: List[PlaylistResponseSchema]) -> List[Playlist]: result = [] for playlist in playlists: playlist_model = self.__get_playlist(playlist.id) if not playlist_model: playlist_model = self.__create_playlist(playlist) db.session.add(playlist_model) else: playlist_model = self.__update_playlist(playlist_model, playlist) result.append(playlist_model) db.session.flush() return result