| 1274 | |
| 1275 | |
| 1276 | class UpdateQuery(AwaitableQuery): |
| 1277 | __slots__ = ( |
| 1278 | "update_kwargs", |
| 1279 | "_orderings", |
| 1280 | "_limit", |
| 1281 | "values", |
| 1282 | ) |
| 1283 | |
| 1284 | def __init__( |
| 1285 | self, |
| 1286 | model: type[MODEL], |
| 1287 | update_kwargs: dict[str, Any], |
| 1288 | db: BaseDBAsyncClient, |
| 1289 | q_objects: list[Q], |
| 1290 | annotations: dict[str, Any], |
| 1291 | custom_filters: dict[str, FilterInfoDict], |
| 1292 | limit: int | None, |
| 1293 | orderings: list[tuple[str, str]], |
| 1294 | ) -> None: |
| 1295 | super().__init__(model) |
| 1296 | self.update_kwargs = update_kwargs |
| 1297 | self._q_objects = q_objects |
| 1298 | self._annotations = annotations |
| 1299 | self._custom_filters = custom_filters |
| 1300 | self._db = db |
| 1301 | self._limit = limit |
| 1302 | self._orderings = orderings |
| 1303 | |
| 1304 | def _make_query(self) -> None: |
| 1305 | table = self.model._meta.basetable |
| 1306 | self.query = self._db.query_class.update(table) |
| 1307 | if self.capabilities.support_update_limit_order_by and self._limit: |
| 1308 | self.query._limit = self.query._wrapper_cls(self._limit) |
| 1309 | self.resolve_ordering(self.model, table, self._orderings, self._annotations) |
| 1310 | |
| 1311 | self.resolve_filters() |
| 1312 | for key, value in self.update_kwargs.items(): |
| 1313 | field_object = self.model._meta.fields_map.get(key) |
| 1314 | if not field_object: |
| 1315 | raise FieldError(f"Unknown keyword argument {key} for model {self.model}") |
| 1316 | if field_object.pk: |
| 1317 | raise IntegrityError(f"Field {key} is PK and can not be updated") |
| 1318 | if field_object.generated: |
| 1319 | raise IntegrityError(f"Field {key} is generated and can not be updated") |
| 1320 | if isinstance(field_object, (ForeignKeyFieldInstance, OneToOneFieldInstance)): |
| 1321 | self.model._validate_relation_type(key, value) |
| 1322 | fk_field: str = field_object.source_field # type: ignore |
| 1323 | db_field = self.model._meta.fields_map[fk_field].source_field |
| 1324 | value = self.model._meta.fields_map[fk_field].to_db_value( |
| 1325 | getattr(value, field_object.to_field_instance.model_field_name), |
| 1326 | None, |
| 1327 | ) |
| 1328 | else: |
| 1329 | try: |
| 1330 | db_field = self.model._meta.fields_db_projection[key] |
| 1331 | except KeyError: |
| 1332 | raise FieldError(f"Field {key} is virtual and can not be updated") |
| 1333 |
no outgoing calls
no test coverage detected
searching dependent graphs…