(self, session: AsyncSession, content_item: Dict)
| 161 | session.add(note) |
| 162 | |
| 163 | async def update_content(self, session: AsyncSession, content_item: Dict): |
| 164 | note_id = content_item.get("note_id") |
| 165 | last_modify_ts = int(get_current_timestamp()) |
| 166 | update_data = { |
| 167 | "last_modify_ts": last_modify_ts, |
| 168 | "liked_count": str(content_item.get("liked_count")), |
| 169 | "collected_count": str(content_item.get("collected_count")), |
| 170 | "comment_count": str(content_item.get("comment_count")), |
| 171 | "share_count": str(content_item.get("share_count")), |
| 172 | "last_update_time": content_item.get("last_update_time"), |
| 173 | } |
| 174 | stmt = update(XhsNote).where(XhsNote.note_id == note_id).values(**update_data) |
| 175 | await session.execute(stmt) |
| 176 | |
| 177 | async def content_is_exist(self, session: AsyncSession, note_id: str) -> bool: |
| 178 | stmt = select(XhsNote).where(XhsNote.note_id == note_id) |
no test coverage detected