(
self,
action: str,
target_type: str,
target_id: Optional[Any] = None,
target_name: str = "",
count: int = 1,
meta: Optional[dict[str, Any]] = None,
)
| 793 | } |
| 794 | |
| 795 | async def record_admin_activity( |
| 796 | self, |
| 797 | action: str, |
| 798 | target_type: str, |
| 799 | target_id: Optional[Any] = None, |
| 800 | target_name: str = "", |
| 801 | count: int = 1, |
| 802 | meta: Optional[dict[str, Any]] = None, |
| 803 | ) -> Optional[dict[str, Any]]: |
| 804 | try: |
| 805 | now = await get_now() |
| 806 | created_at = now.isoformat() |
| 807 | activity = self._normalize_admin_activity( |
| 808 | { |
| 809 | "id": self._build_admin_activity_id( |
| 810 | action=action, |
| 811 | target_type=target_type, |
| 812 | target_id=target_id, |
| 813 | target_name=target_name, |
| 814 | timestamp=now, |
| 815 | ), |
| 816 | "action": action, |
| 817 | "targetType": target_type, |
| 818 | "target_type": target_type, |
| 819 | "targetId": target_id, |
| 820 | "target_id": target_id, |
| 821 | "targetName": target_name, |
| 822 | "target_name": target_name, |
| 823 | "count": count, |
| 824 | "meta": meta or {}, |
| 825 | "createdAt": created_at, |
| 826 | "created_at": created_at, |
| 827 | } |
| 828 | ) |
| 829 | if not activity: |
| 830 | return None |
| 831 | |
| 832 | activities = await self._get_admin_activities() |
| 833 | next_activities = [ |
| 834 | activity, |
| 835 | *[item for item in activities if item["id"] != activity["id"]], |
| 836 | ][: self.MAX_ADMIN_ACTIVITIES] |
| 837 | await self._save_admin_activities(next_activities) |
| 838 | return activity |
| 839 | except Exception: |
| 840 | return None |
| 841 | |
| 842 | async def _get_admin_activities(self) -> list[dict[str, Any]]: |
| 843 | record = await KeyValue.filter(key=self.ADMIN_ACTIVITY_KEY).first() |
no test coverage detected