Take potentially-updated data and apply it to a set of Model objects. The objects are not written back to the database, so the changes are temporary.
(self, objs, old_data, new_data)
| 291 | os.remove(new.name) |
| 292 | |
| 293 | def apply_data(self, objs, old_data, new_data): |
| 294 | """Take potentially-updated data and apply it to a set of Model |
| 295 | objects. |
| 296 | |
| 297 | The objects are not written back to the database, so the changes |
| 298 | are temporary. |
| 299 | """ |
| 300 | if len(old_data) != len(new_data): |
| 301 | self._log.warning( |
| 302 | "number of objects changed from {} to {}", |
| 303 | len(old_data), |
| 304 | len(new_data), |
| 305 | ) |
| 306 | |
| 307 | obj_by_id = {o.id: o for o in objs} |
| 308 | ignore_fields = self.config["ignore_fields"].as_str_seq() |
| 309 | for old_dict, new_dict in zip(old_data, new_data): |
| 310 | # Prohibit any changes to forbidden fields to avoid |
| 311 | # clobbering `id` and such by mistake. |
| 312 | forbidden = False |
| 313 | for key in ignore_fields: |
| 314 | if old_dict.get(key) != new_dict.get(key): |
| 315 | self._log.warning("ignoring object whose {} changed", key) |
| 316 | forbidden = True |
| 317 | break |
| 318 | if forbidden: |
| 319 | continue |
| 320 | |
| 321 | id_ = int(old_dict["id"]) |
| 322 | apply_(obj_by_id[id_], new_dict) |
| 323 | |
| 324 | def save_changes(self, objs): |
| 325 | """Save a list of updated Model objects to the database.""" |
no test coverage detected