youtube_prod_mysql/service/VideoService.py

from entity.VideoEntity import Video
from common.Utils import getSession
from sqlalchemy import update


class VideoService:

    def queryOneByVideoId(videoId):
        session = getSession()
        video = session.query(Video).filter(
            Video.videoId == videoId).one_or_none()
        session.close()
        return video

    def queryAllbyChannelId(channelId):
        session = getSession()
        videos = session.query(Video).filter(
            Video.channelId == channelId).all()
        session.close()
        return videos

    def insertOne(videoId, ChannelId, videoTitle, videoLen, videoType, videoPublishTime, videoLanguage, isDownload):
        session = getSession()
        video: Video = Video(videoId=videoId, ChannelId=ChannelId, videoTitle=videoTitle,
                             videoLen=videoLen, videoType=videoType, videoPublishTime=videoPublishTime,
                             videoLanguage=videoLanguage, isDownload=isDownload)
        session.add(video)
        session.commit()
        session.close()

    def updateLenByVideoId(videoId, videoLen):
        session = getSession()
        updateSql = update(Video).where(
            Video.videoId == videoId).values(videoLen=videoLen)
        resutl = session.execute(updateSql)
        session.commit()
        session.close()

    def getLastVideoByChannelId(channelId):
        session = getSession()
        video: Video = session.query(Video).where(
            Video.channelId == channelId).order_by(Video.videoPublishTime.desc()).first()
        session.close()
        return video

    def upIsDownloadByVideoId(videoId, isDownload):
        session = getSession()
        updateSql = update(Video).where(
            Video.videoId == videoId).values(isDownload=isDownload)
        resutl = session.execute(updateSql)
        session.commit()
        session.close()