(
self, instance: type[Model] | Model, update_fields: Iterable[str] | None
)
| 424 | return sql |
| 425 | |
| 426 | async def execute_update( |
| 427 | self, instance: type[Model] | Model, update_fields: Iterable[str] | None |
| 428 | ) -> int: |
| 429 | |
| 430 | user_specified = update_fields is not None |
| 431 | source_fields: list[str] = ( |
| 432 | list(update_fields) |
| 433 | if update_fields is not None |
| 434 | else list(self.model._meta.fields_db_projection.keys()) |
| 435 | ) |
| 436 | |
| 437 | effective_fields: list[str] = [] |
| 438 | for field in source_fields: |
| 439 | field_obj = self.model._meta.fields_map[field] |
| 440 | if field_obj.pk: |
| 441 | if user_specified: |
| 442 | raise OperationalError( |
| 443 | f"Can't update pk field, use `{self.model.__name__}.create` instead." |
| 444 | ) |
| 445 | continue |
| 446 | if field_obj.generated: |
| 447 | if user_specified: |
| 448 | raise OperationalError(f"Can't update generated field {field}") |
| 449 | continue |
| 450 | instance_field = getattr(instance, field) |
| 451 | if isinstance(instance_field, DatabaseDefault): |
| 452 | continue |
| 453 | effective_fields.append(field) |
| 454 | |
| 455 | if not effective_fields: |
| 456 | return 0 |
| 457 | |
| 458 | values = [] |
| 459 | expressions = {} |
| 460 | for field in effective_fields: |
| 461 | instance_field = getattr(instance, field) |
| 462 | if isinstance(instance_field, Expression): |
| 463 | expressions[field] = instance_field |
| 464 | else: |
| 465 | field_obj = self.model._meta.fields_map[field] |
| 466 | values.append(field_obj.to_db_value(instance_field, instance)) |
| 467 | |
| 468 | values.append(self.model._meta.pk.to_db_value(instance.pk, instance)) |
| 469 | return ( |
| 470 | await self.db.execute_query(self.get_update_sql(effective_fields, expressions), values) |
| 471 | )[0] |
| 472 | |
| 473 | async def execute_delete(self, instance: type[Model] | Model) -> int: |
| 474 | return ( |
no test coverage detected