LongArticleTaskServer/applications/utils/async_mysql_utils.py

from typing import List, Dict


async def get_top_article_title_list(pool) -> List[Dict]:
    query = """
        select distinct title, source_id
        from datastat_sort_strategy
        where produce_plan_name = %s and source_id is not null;
    """
    return await pool.async_fetch(query=query, params=("TOP100",))


async def get_hot_titles(
    pool, date_string, position, read_times_threshold
) -> List[str]:
    """get titles of hot articles"""
    query = """
        select distinct title
        from datastat_sort_strategy
        where position < %s and read_rate >= %s and date_str >= %s;
    """
    response = await pool.async_fetch(
        query=query, params=(position, read_times_threshold, date_string)
    )
    return [i["title"] for i in response]