(self)
| 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 | |
| 1334 | if isinstance(value, Expression): |
| 1335 | value = value.resolve( |
| 1336 | ResolveContext( |
| 1337 | model=self.model, |
| 1338 | table=table, |
| 1339 | annotations=self._annotations, |
| 1340 | custom_filters=self._custom_filters, |
| 1341 | ) |
| 1342 | ).term |
| 1343 | else: |
| 1344 | value = self.model._meta.fields_map[key].to_db_value(value, None) |
| 1345 | |
| 1346 | self.query = self.query.set(db_field, value) |
| 1347 | |
| 1348 | def __await__(self) -> Generator[Any, None, int]: |
| 1349 | self._choose_db_if_not_chosen(True) |
no test coverage detected