| 1234567891011121314151617181920212223242526272829303132333435363738 |
- from typing import Any
- from app.api.v1.module_system.auth.schema import AuthSchema
- from app.core.base_crud import CRUDBase
- from app.core.exceptions import CustomException
- from .model import AlipayNotifyLogModel
- class AlipayNotifyLogCRUD(CRUDBase[AlipayNotifyLogModel, Any, Any]):
- """支付宝通知日志 CRUD 操作"""
- def __init__(self, auth: AuthSchema) -> None:
- self.auth = auth
- super().__init__(model=AlipayNotifyLogModel, auth=auth)
- async def get_by_notify_id(
- self, notify_id: str
- ) -> AlipayNotifyLogModel | None:
- return await self.get(notify_id=notify_id)
- async def update_by_notify_id(
- self, notify_id: str, data: dict
- ) -> AlipayNotifyLogModel | None:
- obj = await self.get(notify_id=notify_id, preload=[])
- if not obj:
- raise CustomException(msg="通知日志不存在")
- if self.auth.user and hasattr(obj, "updated_id"):
- setattr(obj, "updated_id", self.auth.user.id)
- for key, value in data.items():
- if hasattr(obj, key):
- setattr(obj, key, value)
- await self.auth.db.flush()
- await self.auth.db.refresh(obj)
- return obj
|
