(
self,
uid: Any,
data: dict[str, Any] | T,
)
| 150 | return entity |
| 151 | |
| 152 | async def update( |
| 153 | self, |
| 154 | uid: Any, |
| 155 | data: dict[str, Any] | T, |
| 156 | ) -> T: |
| 157 | if isinstance(data, dict): |
| 158 | d = data |
| 159 | else: |
| 160 | d = data.__dict__ |
| 161 | |
| 162 | entity = await self.get_by_id(uid) |
| 163 | |
| 164 | for field, value in d.items(): |
| 165 | if hasattr(entity, field): |
| 166 | setattr(entity, field, value) |
| 167 | async with self.get_session() as session: |
| 168 | session.add(entity) |
| 169 | await session.flush() |
| 170 | await session.refresh(entity) |
| 171 | return entity |
| 172 | |
| 173 | async def update_many(self, filters: list[Filter], update_data: dict[str, Any]) -> int: |
| 174 | query = update(self._model) |
no test coverage detected