LongArticleTaskServer/app/infra/mapper/long_video_mapper.py

from typing import List, Dict

from app.core.database import DatabaseManager


class LongVideoDatabaseMapper:
    # 从 long_video 数据库查询 video_cover_snapshots 信息
    @staticmethod
    async def fetch_long_video_cover(
        pool: DatabaseManager, video_id: str
    ) -> List[Dict]:
        """
        use channel_content_id to find long video cover
        """
        query = """
            SELECT image_path FROM video_cover_snapshots WHERE video_id = %s;
        """
        return await pool.async_fetch(
            query=query, db_name="long_video", params=(video_id,)
        )