(self, session: AsyncSession, comment_item: Dict)
| 211 | session.add(comment) |
| 212 | |
| 213 | async def update_comment(self, session: AsyncSession, comment_item: Dict): |
| 214 | comment_id = comment_item.get("comment_id") |
| 215 | last_modify_ts = int(get_current_timestamp()) |
| 216 | update_data = { |
| 217 | "last_modify_ts": last_modify_ts, |
| 218 | "like_count": str(comment_item.get("like_count")), |
| 219 | "sub_comment_count": int(comment_item.get("sub_comment_count", 0) or 0), |
| 220 | } |
| 221 | stmt = update(XhsNoteComment).where(XhsNoteComment.comment_id == comment_id).values(**update_data) |
| 222 | await session.execute(stmt) |
| 223 | |
| 224 | async def comment_is_exist(self, session: AsyncSession, comment_id: str) -> bool: |
| 225 | stmt = select(XhsNoteComment).where(XhsNoteComment.comment_id == comment_id) |
no test coverage detected