1.cache/news_cache.py
NEWS_LIST_PREFIX = "news_list:" # 写入缓存-新闻列表 key = news_list:分类id:页码:每页数量 + 列表数据 + 过期时间 async def set_cache_news_list(category_id: Optional[int], page: int, size: int, news_list: List[Dict[str, Any]], expire: int = 1800): category_part = category_id if category_id is not None else "all" key = f"{NEWS_LIST_PREFIX}{category_part}:{page}:{size}" return await set_cache(key, news_list, expire) # 读取缓存-新闻列表 key = news_list:分类id:页码:每页数量 async def get_cache_news_list(category_id: Optional[int], page: int, size: int): category_part = category_id if category_id is not None else "all" key = f"{NEWS_LIST_PREFIX}{category_part}:{page}:{size}" return await get_json_cache(key)
2. 改造crud/news.py
# 获取新闻列表 async def get_news_list( db: AsyncSession, category_id: int, skip: int = 0, limit: int = 10 ): # 先尝试从缓存获取新闻列表 # 跳过的数量skip = (页码 - 1) * 每页数量 -> 页码 = 跳过的数量 // 每页数量 + 1 # await get_cache_news_list(分类id, 页码, 每页数量) page = skip // limit + 1 cached_list = await get_cache_news_list(category_id, page, limit) # 缓存数据 json if cached_list: # return cached_list # 要的是 ORM return [News(**item) for item in cached_list] # 查询的是指定分类下的所有新闻 stmt = select(News).where(News.category_id == category_id).offset(skip).limit(limit) result = await db.execute(stmt) news_list = result.scalars().all() # 写入缓存 if news_list: # 先把 ORM 数据 转换 字典才能写入缓存 # ORM 转成 Pydantic，再转为 字典 # by_alias=False 不适用别名，保存 Python 风格，因为 Redis 数据是给后端用的 news_data = [ NewsItemBase.model_validate(item).model_dump(mode="json", by_alias=False) for item in news_list ] await set_cache_news_list(category_id, page, limit, news_data) return news_list
