Coverage for src/backend/InvenTree/importer/models.py: 16%

1"""Model definitions for the 'importer' app.""" 

2 

3import json 

4from collections import OrderedDict 

5from datetime import datetime 

6from typing import Optional 

7 

8from django.contrib.auth.models import User 

9from django.core.exceptions import FieldDoesNotExist 

10from django.core.exceptions import ValidationError as DjangoValidationError 

11from django.core.validators import FileExtensionValidator 

12from django.db import models 

13from django.urls import reverse 

14from django.utils.translation import gettext_lazy as _ 

15 

16import structlog 

17from rest_framework.exceptions import ValidationError as DRFValidationError 

18 

19import importer.operations 

20import importer.registry 

21import importer.tasks 

22import importer.validators 

23import InvenTree.helpers 

24from common.models import RenderChoices 

25from importer.status_codes import DataImportStatusCode 

26 

27logger = structlog.get_logger('inventree') 

28 

29 

30class DataImportSession(models.Model): 

31 """Database model representing a data import session. 

32 

33 An initial file is uploaded, and used to populate the database. 

34 

35 Fields: 

36 timestamp: Timestamp for the import session 

37 data_file: FileField for the data file to import 

38 status: IntegerField for the status of the import session 

39 user: ForeignKey to the User who initiated the import 

40 field_defaults: JSONField for field default values - provides a backup value for a field 

41 field_overrides: JSONField for field override values - used to force a value for a field 

42 field_filters: JSONField for field filter values - optional field API filters 

43 """ 

44 

45 ID_FIELD_LABEL = 'id' 

46 

47 class ModelChoices(RenderChoices): 

48 """Model choices for data import sessions.""" 

49 

50 choice_fnc = importer.registry.supported_models 

51 

52 @staticmethod 

53 def get_api_url(): 

54 """Return the API URL associated with the DataImportSession model.""" 

55 return reverse('api-importer-session-list') 

56 

57 def save(self, *args, **kwargs): 

58 """Save the DataImportSession object.""" 

59 initial = self.pk is None 

60 

61 self.clean() 

62 

63 super().save(*args, **kwargs) 

64 

65 if initial: 

66 # New object - run initial setup 

67 self.status = DataImportStatusCode.INITIAL.value 

68 self.progress = 0 

69 self.extract_columns() 

70 

71 timestamp = models.DateTimeField(auto_now_add=True, verbose_name=_('Timestamp')) 

72 

73 data_file = models.FileField( 

74 upload_to='import', 

75 verbose_name=_('Data File'), 

76 help_text=_('Data file to import'), 

77 validators=[ 

78 FileExtensionValidator( 

79 allowed_extensions=InvenTree.helpers.GetExportFormats() 

80 ), 

81 importer.validators.validate_data_file, 

82 ], 

83 ) 

84 

85 columns = models.JSONField(blank=True, null=True, verbose_name=_('Columns')) 

86 

87 model_type = models.CharField( 

88 blank=False, 

89 max_length=100, 

90 validators=[importer.validators.validate_importer_model_type], 

91 verbose_name=_('Model Type'), 

92 help_text=_('Target model type for this import session'), 

93 ) 

94 

95 status = models.PositiveIntegerField( 

96 default=DataImportStatusCode.INITIAL.value, 

97 choices=DataImportStatusCode.items(), 

98 help_text=_('Import status'), 

99 ) 

100 

101 user = models.ForeignKey( 

102 User, on_delete=models.SET_NULL, blank=True, null=True, verbose_name=_('User') 

103 ) 

104 

105 field_defaults = models.JSONField( 

106 blank=True, 

107 null=True, 

108 verbose_name=_('Field Defaults'), 

109 validators=[importer.validators.validate_field_defaults], 

110 ) 

111 

112 field_overrides = models.JSONField( 

113 blank=True, 

114 null=True, 

115 verbose_name=_('Field Overrides'), 

116 validators=[importer.validators.validate_field_defaults], 

117 ) 

118 

119 field_filters = models.JSONField( 

120 blank=True, 

121 null=True, 

122 verbose_name=_('Field Filters'), 

123 validators=[importer.validators.validate_field_defaults], 

124 ) 

125 

126 update_records = models.BooleanField( 

127 default=False, 

128 verbose_name=_('Update Existing Records'), 

129 help_text=_('If enabled, existing records will be updated with new data'), 

130 ) 

131 

132 @property 

133 def field_mapping(self) -> dict: 

134 """Construct a dict of field mappings for this import session. 

135 

136 Returns: 

137 A dict of field -> column mappings 

138 """ 

139 return {mapping.field: mapping.column for mapping in self.column_mappings.all()} 

140 

141 @property 

142 def model_class(self): 

143 """Return the model class for this importer.""" 

144 serializer = self.serializer_class 

145 

146 if serializer: 

147 return serializer.Meta.model 

148 

149 @property 

150 def serializer_class(self): 

151 """Return the serializer class for this importer.""" 

152 from importer.registry import supported_models 

153 

154 return supported_models().get(self.model_type, None) 

155 

156 def get_lookup_fields_for_field(self, field_name: str) -> list: 

157 """Return the valid lookup fields for a given related (FK) field. 

158 

159 Returns a list of field names that can be used as a lookup key, 

160 consisting of 'pk' plus any fields defined in IMPORT_ID_FIELDS on the related model. 

161 """ 

162 model = self.get_related_model(field_name) 

163 

164 if not model: 

165 return ['pk'] 

166 

167 id_fields = ['pk'] 

168 

169 if custom_fields := getattr(model, 'IMPORT_ID_FIELDS', None): 

170 id_fields += custom_fields 

171 

172 return id_fields 

173 

174 @property 

175 def field_lookup_mapping(self) -> dict: 

176 """Return a dict of field -> lookup_field mappings for this import session. 

177 

178 Only entries where lookup_field is explicitly set are included. 

179 """ 

180 return { 

181 mapping.field: mapping.lookup_field 

182 for mapping in self.column_mappings.all() 

183 if mapping.lookup_field 

184 } 

185 

186 def get_related_model(self, field_name: str) -> Optional[models.Model]: 

187 """Return the related model for a given field name. 

188 

189 Arguments: 

190 field_name: The name of the field to check 

191 

192 Returns: 

193 The related model class, if one exists, or None otherwise 

194 """ 

195 model_class = self.model_class 

196 

197 if not model_class: 

198 return None 

199 

200 try: 

201 related_field = model_class._meta.get_field(field_name) 

202 model = related_field.remote_field.model 

203 return model 

204 except (AttributeError, FieldDoesNotExist): 

205 return None 

206 

207 def extract_columns(self) -> None: 

208 """Run initial column extraction and mapping. 

209 

210 This method is called when the import session is first created. 

211 

212 - Extract column names from the data file 

213 - Create a default mapping for each field in the serializer 

214 - Find a default "backup" value for each field (if one exists) 

215 """ 

216 # Extract list of column names from the file 

217 self.columns = importer.operations.extract_column_names(self.data_file) 

218 

219 serializer_fields = self.available_fields() 

220 

221 # Remove any existing mappings 

222 self.column_mappings.all().delete() 

223 

224 column_mappings = [] 

225 

226 matched_columns = set() 

227 

228 self.field_defaults = self.field_defaults or {} 

229 field_overrides = self.field_overrides or {} 

230 

231 # Create a default mapping for each available field in the database 

232 for field, field_def in serializer_fields.items(): 

233 # If an override value is provided for the field, 

234 # skip creating a mapping for this field 

235 if field in field_overrides: 

236 continue 

237 

238 # Extract a "default" value for the field, if one exists 

239 # Skip if one has already been provided by the user 

240 if field not in self.field_defaults and 'default' in field_def: 

241 self.field_defaults[field] = field_def['default'] 

242 

243 # Generate a list of possible column names for this field 

244 field_options = [ 

245 field, 

246 field_def.get('label', field), 

247 field_def.get('help_text', field), 

248 ] 

249 column_name = '' 

250 

251 for column in self.columns: 

252 # No title provided for the column 

253 if not column: 

254 continue 

255 

256 # Ignore if we have already matched this column to a field 

257 if column in matched_columns: 

258 continue 

259 

260 # Try direct match 

261 if column in field_options: 

262 column_name = column 

263 break 

264 

265 # Try lower case match 

266 if column.lower() in [f.lower() for f in field_options]: 

267 column_name = column 

268 break 

269 

270 column_mappings.append( 

271 DataImportColumnMap(session=self, column=column_name, field=field) 

272 ) 

273 

274 # Create the column mappings 

275 DataImportColumnMap.objects.bulk_create(column_mappings, batch_size=250) 

276 

277 self.status = DataImportStatusCode.MAPPING.value 

278 self.save() 

279 

280 def accept_mapping(self) -> None: 

281 """Accept current mapping configuration. 

282 

283 - Validate that the current column mapping is correct 

284 - Trigger the data import process 

285 """ 

286 # First, we need to ensure that all the *required* columns have been mapped 

287 required_fields = self.required_fields() 

288 

289 field_defaults = self.field_defaults or {} 

290 field_overrides = self.field_overrides or {} 

291 

292 missing_fields = [] 

293 

294 for field in required_fields: 

295 # An override value exists 

296 if field in field_overrides: 

297 continue 

298 

299 # A default value exists 

300 if field_defaults.get(field): 

301 continue 

302 

303 # The field has been mapped to a data column 

304 if mapping := self.column_mappings.filter(field=field).first(): 

305 if mapping.column: 

306 continue 

307 

308 missing_fields.append(field) 

309 

310 if len(missing_fields) > 0: 

311 raise DjangoValidationError({ 

312 'error': _('Some required fields have not been mapped'), 

313 'fields': missing_fields, 

314 }) 

315 

316 # No errors, so trigger the data import process 

317 self.trigger_data_import() 

318 

319 def trigger_data_import(self) -> None: 

320 """Trigger the data import process for this session. 

321 

322 Offloads the task to the background worker process. 

323 """ 

324 from InvenTree.tasks import offload_task 

325 

326 # Mark the import task status as "IMPORTING" 

327 self.status = DataImportStatusCode.IMPORTING.value 

328 self.save() 

329 

330 offload_task(importer.tasks.import_data, self.pk, group='importer') 

331 

332 def import_data(self) -> None: 

333 """Perform the data import process for this session.""" 

334 # Clear any existing data rows 

335 self.rows.all().delete() 

336 

337 df = importer.operations.load_data_file(self.data_file) 

338 

339 if df is None: 

340 # TODO: Log an error message against the import session 

341 logger.error('Failed to load data file') 

342 return 

343 

344 headers = importer.operations.normalize_headers(df.headers) 

345 

346 imported_rows = [] 

347 

348 field_mapping = self.field_mapping 

349 available_fields = self.available_fields() 

350 

351 # Iterate through each "row" in the data file, and create a new DataImportRow object 

352 for idx, row in enumerate(df): 

353 row_data = dict(zip(headers, row, strict=False)) 

354 

355 # Skip completely empty rows 

356 if not any(row_data.values()): 

357 continue 

358 

359 row = DataImportRow(session=self, row_data=row_data, row_index=idx) 

360 

361 row.extract_data( 

362 field_mapping=field_mapping, 

363 available_fields=available_fields, 

364 commit=False, 

365 ) 

366 

367 row.valid = row.validate(commit=False) 

368 imported_rows.append(row) 

369 

370 # Perform database writes as a single operation 

371 DataImportRow.objects.bulk_create(imported_rows, batch_size=250) 

372 

373 # Mark the import task as "PROCESSING" 

374 self.status = DataImportStatusCode.PROCESSING.value 

375 self.save() 

376 

377 def check_complete(self) -> bool: 

378 """Check if the import session is complete. 

379 

380 When all rows have been accepted, the rows and column mappings are 

381 deleted as they are no longer needed. The session itself is retained 

382 as an audit record. 

383 """ 

384 if self.completed_row_count < self.row_count: 

385 return False 

386 

387 if self.status != DataImportStatusCode.COMPLETE.value: 

388 self.status = DataImportStatusCode.COMPLETE.value 

389 

390 # persist historic count values for reporting purposes 

391 self.completed_row_count_history = self.completed_row_count 

392 self.row_count_history = self.row_count 

393 

394 self.save() 

395 

396 # Clear staging data now that all rows have been imported 

397 self.rows.all().delete() 

398 self.column_mappings.all().delete() 

399 

400 return True 

401 

402 @property 

403 def row_count(self) -> int: 

404 """Return the number of rows in the import session.""" 

405 return self.rows.count() 

406 

407 @property 

408 def completed_row_count(self) -> int: 

409 """Return the number of completed rows for this session.""" 

410 return self.rows.filter(complete=True).count() 

411 

412 # Historic values for reporting purposes 

413 completed_row_count_history = models.PositiveIntegerField( 

414 blank=True, null=True, verbose_name=_('Completed Row Count History') 

415 ) 

416 row_count_history = models.PositiveIntegerField( 

417 blank=True, null=True, verbose_name=_('Row Count History') 

418 ) 

419 

420 def available_fields(self): 

421 """Returns information on the available fields. 

422 

423 - This method is designed to be introspected by the frontend, for rendering the various fields. 

424 - We make use of the InvenTree.metadata module to provide extra information about the fields. 

425 

426 Note that we cache these fields, as they are expensive to compute. 

427 """ 

428 if fields := getattr(self, '_available_fields', None): 

429 return fields 

430 

431 from InvenTree.metadata import InvenTreeMetadata 

432 

433 metadata = InvenTreeMetadata() 

434 

435 fields = OrderedDict() 

436 

437 if self.update_records: 

438 # If we are updating records, ensure the ID field is included 

439 fields[self.ID_FIELD_LABEL] = { 

440 'label': _('ID'), 

441 'help_text': _('Existing database identifier for the record'), 

442 'type': 'integer', 

443 'required': True, 

444 'read_only': False, 

445 } 

446 

447 if serializer_class := self.serializer_class: 

448 serializer = serializer_class(data={}, importing=True) 

449 serializer_fields = metadata.get_serializer_info(serializer) 

450 

451 for field_name, field in serializer_fields.items(): 

452 # Skip read-only fields 

453 if field.get('read_only', False): 

454 continue 

455 

456 if field.get('type') == 'related field': 

457 field['lookup_fields'] = self.get_lookup_fields_for_field( 

458 field_name 

459 ) 

460 

461 fields[field_name] = field 

462 

463 # Cache the available fields against this instance 

464 self._available_fields = fields 

465 

466 return fields 

467 

468 def required_fields(self) -> dict: 

469 """Returns information on which fields are *required* for import.""" 

470 fields = self.available_fields() 

471 

472 required = {} 

473 

474 for field, info in fields.items(): 

475 if info.get('required', False): 

476 required[field] = info 

477 

478 elif self.update_records and field == self.ID_FIELD_LABEL: 

479 # If we are updating records, the ID field is required 

480 required[field] = info 

481 

482 return required 

483 

484 

485class DataImportColumnMap(models.Model): 

486 """Database model representing a mapping between a file column and serializer field. 

487 

488 - Each row maps a "column" (in the import file) to a "field" (in the serializer) 

489 - Column must exist in the file 

490 - Field must exist in the serializer (and not be read-only) 

491 """ 

492 

493 @staticmethod 

494 def get_api_url(): 

495 """Return the API URL associated with the DataImportColumnMap model.""" 

496 return reverse('api-importer-mapping-list') 

497 

498 def save(self, *args, **kwargs): 

499 """Save the DataImportColumnMap object.""" 

500 self.clean() 

501 self.validate_unique() 

502 

503 super().save(*args, **kwargs) 

504 

505 def validate_unique(self, exclude=None): 

506 """Ensure that the column mapping is unique within the session.""" 

507 super().validate_unique(exclude) 

508 

509 columns = self.session.column_mappings.exclude(pk=self.pk) 

510 

511 if ( 

512 self.column not in ['', None] 

513 and columns.filter(column=self.column).exists() 

514 ): 

515 raise DjangoValidationError({ 

516 'column': _('Column is already mapped to a database field') 

517 }) 

518 

519 if columns.filter(field=self.field).exists(): 

520 raise DjangoValidationError({ 

521 'field': _('Field is already mapped to a data column') 

522 }) 

523 

524 def clean(self): 

525 """Validate the column mapping.""" 

526 super().clean() 

527 

528 if not self.session: 

529 raise DjangoValidationError({ 

530 'session': _('Column mapping must be linked to a valid import session') 

531 }) 

532 

533 if self.column and self.column not in self.session.columns: 

534 raise DjangoValidationError({ 

535 'column': _('Column does not exist in the data file') 

536 }) 

537 

538 field_def = self.field_definition 

539 

540 if not field_def: 

541 raise DjangoValidationError({ 

542 'field': _('Field does not exist in the target model') 

543 }) 

544 

545 if field_def.get('read_only', False): 

546 raise DjangoValidationError({'field': _('Selected field is read-only')}) 

547 

548 if self.lookup_field: 

549 if field_def.get('type') != 'related field': 

550 raise DjangoValidationError({ 

551 'lookup_field': _( 

552 'Lookup field can only be set for related (foreign-key) fields' 

553 ) 

554 }) 

555 

556 valid_lookup_fields = self.session.get_lookup_fields_for_field(self.field) 

557 if self.lookup_field not in valid_lookup_fields: 

558 raise DjangoValidationError({ 

559 'lookup_field': _( 

560 'Invalid lookup field. Valid options are: {options}' 

561 ).format(options=', '.join(valid_lookup_fields)) 

562 }) 

563 

564 session = models.ForeignKey( 

565 DataImportSession, 

566 on_delete=models.CASCADE, 

567 verbose_name=_('Import Session'), 

568 related_name='column_mappings', 

569 ) 

570 

571 field = models.CharField(max_length=100, verbose_name=_('Field')) 

572 

573 column = models.CharField(blank=True, max_length=100, verbose_name=_('Column')) 

574 

575 lookup_field = models.CharField( 

576 blank=True, 

577 null=True, 

578 max_length=100, 

579 verbose_name=_('Lookup Field'), 

580 help_text=_( 

581 'Database field to use for foreign-key lookup. Leave blank for automatic lookup.' 

582 ), 

583 ) 

584 

585 @property 

586 def available_fields(self): 

587 """Return a list of available fields for this import session. 

588 

589 These fields get cached, as they are expensive to compute. 

590 """ 

591 if fields := getattr(self, '_available_fields', None): 

592 return fields 

593 

594 self._available_fields = self.session.available_fields() 

595 

596 return self._available_fields 

597 

598 @property 

599 def field_definition(self): 

600 """Return the field definition associated with this column mapping.""" 

601 fields = self.available_fields 

602 return fields.get(self.field, None) 

603 

604 @property 

605 def label(self): 

606 """Extract the 'label' associated with the mapped field.""" 

607 if field_def := self.field_definition: 

608 return field_def.get('label', None) 

609 

610 @property 

611 def description(self): 

612 """Extract the 'description' associated with the mapped field.""" 

613 description = None 

614 

615 if field_def := self.field_definition: 

616 description = field_def.get('help_text', None) 

617 

618 if not description: 

619 description = self.label 

620 

621 return description 

622 

623 

624class DataImportRow(models.Model): 

625 """Database model representing a single row in a data import session. 

626 

627 Each row corresponds to a single row in the import file, and is used to populate the database. 

628 

629 Fields: 

630 session: ForeignKey to the parent DataImportSession object 

631 data: JSONField for the data in this row 

632 status: IntegerField for the status of the row import 

633 """ 

634 

635 @staticmethod 

636 def get_api_url(): 

637 """Return the API URL associated with the DataImportRow model.""" 

638 return reverse('api-importer-row-list') 

639 

640 def save(self, *args, **kwargs): 

641 """Save the DataImportRow object.""" 

642 self.valid = self.validate() 

643 super().save(*args, **kwargs) 

644 

645 def delete(self, *args, **kwargs): 

646 """Update the session progress when a row is deleted.""" 

647 session = self.session 

648 super().delete(*args, **kwargs) 

649 session.check_complete() 

650 

651 session = models.ForeignKey( 

652 DataImportSession, 

653 on_delete=models.CASCADE, 

654 verbose_name=_('Import Session'), 

655 related_name='rows', 

656 ) 

657 

658 row_index = models.PositiveIntegerField(default=0, verbose_name=_('Row Index')) 

659 

660 row_data = models.JSONField( 

661 blank=True, null=True, verbose_name=_('Original row data') 

662 ) 

663 

664 data = models.JSONField(blank=True, null=True, verbose_name=_('Data')) 

665 

666 errors = models.JSONField(blank=True, null=True, verbose_name=_('Errors')) 

667 

668 valid = models.BooleanField(default=False, verbose_name=_('Valid')) 

669 

670 complete = models.BooleanField(default=False, verbose_name=_('Complete')) 

671 

672 @property 

673 def default_values(self) -> dict: 

674 """Return a dict object of the 'default' values for this row.""" 

675 defaults = self.session.field_defaults or {} 

676 

677 if type(defaults) is not dict: 

678 try: 

679 defaults = json.loads(str(defaults)) 

680 except json.JSONDecodeError: 

681 logger.warning('Failed to parse default values for import row') 

682 defaults = {} 

683 

684 return defaults 

685 

686 @property 

687 def override_values(self) -> dict: 

688 """Return a dict object of the 'override' values for this row.""" 

689 overrides = self.session.field_overrides or {} 

690 

691 if type(overrides) is not dict: 

692 try: 

693 overrides = json.loads(str(overrides)) 

694 except json.JSONDecodeError: 

695 logger.warning('Failed to parse override values for import row') 

696 overrides = {} 

697 

698 return overrides 

699 

700 def extract_data( 

701 self, 

702 available_fields: Optional[dict] = None, 

703 field_mapping: Optional[dict] = None, 

704 commit=True, 

705 ): 

706 """Extract row data from the provided data dictionary.""" 

707 if not field_mapping: 

708 field_mapping = self.session.field_mapping 

709 

710 if not available_fields: 

711 available_fields = self.session.available_fields() 

712 

713 override_values = self.override_values 

714 default_values = self.default_values 

715 

716 data = {} 

717 extract_errors = {} 

718 

719 self.related_field_map = {} 

720 

721 field_lookup_mapping = self.session.field_lookup_mapping 

722 

723 # We have mapped column (file) to field (serializer) already 

724 for field, col in field_mapping.items(): 

725 # Data override (force value and skip any further checks) 

726 if field in override_values: 

727 data[field] = override_values[field] 

728 continue 

729 

730 # Default value (if provided) 

731 if field in default_values: 

732 data[field] = default_values[field] 

733 

734 # If this field is *not* mapped to any column, skip 

735 if not col or col not in self.row_data: 

736 continue 

737 

738 # Extract field type 

739 field_def = available_fields.get(field, {}) 

740 

741 field_type = field_def.get('type', None) 

742 

743 value = self.row_data.get(col, None) 

744 

745 if field_type == 'boolean': 

746 value = InvenTree.helpers.str2bool(value) 

747 elif field_type == 'date': 

748 value = self.convert_date_field(value) 

749 elif field_type == 'related field': 

750 try: 

751 value = self.lookup_related_field( 

752 field, value, lookup_field=field_lookup_mapping.get(field) 

753 ) 

754 except DjangoValidationError as exc: 

755 extract_errors[field] = exc.message 

756 continue 

757 

758 # Use the default value, if provided 

759 if value is None and field in default_values: 

760 value = default_values[field] 

761 

762 # If the field provides a set of valid 'choices', use that as a lookup 

763 if field_type == 'choice' and 'choices' in field_def: 

764 choices = field_def.get('choices', None) 

765 

766 if callable(choices): 

767 choices = choices() 

768 

769 # Try to match the provided value against the available choices 

770 choice_value = None 

771 

772 for choice in choices: 

773 primary_value = choice['value'] 

774 display_value = choice['display_name'] 

775 

776 if primary_value == value: 

777 choice_value = primary_value 

778 # Break on first match against a primary choice value 

779 break 

780 

781 if display_value == value: 

782 choice_value = primary_value 

783 

784 elif ( 

785 str(display_value).lower().strip() == str(value).lower().strip() 

786 and choice_value is None 

787 ): 

788 # Case-insensitive match against display value 

789 choice_value = primary_value 

790 

791 if choice_value is not None: 

792 value = choice_value 

793 

794 data[field] = value 

795 

796 self.data = data 

797 

798 if extract_errors: 

799 self.errors = extract_errors 

800 

801 if commit: 

802 self.save() 

803 

804 def convert_date_field(self, value: str) -> Optional[str]: 

805 """Convert an incoming date field to the correct format for the database.""" 

806 if value in [None, '']: 

807 return None 

808 

809 # Attempt conversion using accepted formats 

810 date_formats = ['%Y-%m-%d', '%d/%m/%Y', '%m/%d/%Y', '%Y/%m/%d'] 

811 

812 for fmt in date_formats: 

813 try: 

814 dt = datetime.strptime(value.strip(), fmt) 

815 

816 # If the date is valid, convert it to the standard format and return 

817 return dt.strftime('%Y-%m-%d') 

818 except ValueError: 

819 continue 

820 

821 # If none of the formats matched, return the original value 

822 return value 

823 

824 def lookup_related_field( 

825 self, field_name: str, value: str, lookup_field: Optional[str] = None 

826 ) -> Optional[int]: 

827 """Try to perform lookup against a related field. 

828 

829 - This is used to convert a human-readable value (e.g. a supplier name) into a database reference (e.g. supplier ID). 

830 - Reference the value against the related model's allowable import fields 

831 

832 Arguments: 

833 field_name: The name of the field to perform the lookup against 

834 value: The value to be looked up 

835 lookup_field: If provided, only query this specific model field (skips auto-lookup) 

836 

837 Returns: 

838 A primary key value 

839 """ 

840 if value is None or value == '': 

841 return value 

842 

843 if field_name is None or field_name == '': 

844 return value 

845 

846 if field_name in self.related_field_map: 

847 model = self.related_field_map[field_name] 

848 else: 

849 # Cache the related model for this field name 

850 model = self.related_field_map[field_name] = self.session.get_related_model( 

851 field_name 

852 ) 

853 

854 if not model: 

855 raise DjangoValidationError({ 

856 'session': f'No related model found for field: {field_name}' 

857 }) 

858 

859 base_filters = ( 

860 self.session.field_filters.get(field_name, {}) 

861 if self.session.field_filters 

862 else {} 

863 ) 

864 

865 if lookup_field and type(lookup_field) is str: 

866 # A specific lookup field has been chosen by the user — query only that field 

867 try: 

868 queryset = model.objects.filter(**{lookup_field: value}, **base_filters) 

869 except ValueError: 

870 return value 

871 

872 results = list(queryset[:2]) 

873 

874 if len(results) == 1: 

875 return results[0].pk 

876 

877 # Zero or multiple results — return raw value and let serializer report the error 

878 return value 

879 

880 # Auto-lookup: try pk first, then any model-defined IMPORT_ID_FIELDS 

881 id_fields = ['pk'] 

882 

883 if custom_id_fields := getattr(model, 'IMPORT_ID_FIELDS', None): 

884 id_fields += custom_id_fields 

885 

886 valid_items = set() 

887 

888 for id_field in id_fields: 

889 try: 

890 queryset = model.objects.filter(**{id_field: value}, **base_filters) 

891 except ValueError: 

892 continue 

893 

894 # Evaluate at most two results to determine if there is exactly one match 

895 results = list(queryset[:2]) 

896 if len(results) == 1: 

897 valid_items.add(results[0].pk) 

898 

899 if len(valid_items) == 1: 

900 return valid_items.pop()