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) -> List[Dict]:
    """get titles of hot articles"""
    query = """
        select distinct title
        from datastat_sort_strategy
        where position < %s and read_rate >= %s and date_str >= %s;
    """
    return await pool.async_fetch(query=query, params=(3, 1.21, date_string))