Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/import_export/instance_loaders.py: 29%

1class BaseInstanceLoader: 

2 """ 

3 Base abstract implementation of instance loader. 

4 """ 

5 

6 def __init__(self, resource, dataset=None): 

7 self.resource = resource 

8 self.dataset = dataset 

9 

10 def get_instance(self, row): 

11 raise NotImplementedError 

12 

13 

14class ModelInstanceLoader(BaseInstanceLoader): 

15 """ 

16 Instance loader for Django model. 

17 

18 Lookup for model instance by ``import_id_fields``. 

19 """ 

20 

21 def get_queryset(self): 

22 return self.resource.get_queryset() 

23 

24 def get_instance(self, row): 

25 try: 

26 params = {} 

27 for key in self.resource.get_import_id_fields(): 

28 field = self.resource.fields[key] 

29 params[field.attribute] = field.clean(row) 

30 if params: 

31 return self.get_queryset().get(**params) 

32 else: 

33 return None 

34 except self.resource._meta.model.DoesNotExist: 

35 return None 

36 

37 

38class CachedInstanceLoader(ModelInstanceLoader): 

39 """ 

40 Loads all possible model instances in dataset avoid hitting database for 

41 every ``get_instance`` call. 

42 

43 This instance loader work only when there is one ``import_id_fields`` 

44 field. 

45 """ 

46 

47 def __init__(self, *args, **kwargs): 

48 super().__init__(*args, **kwargs) 

49 

50 pk_field_name = self.resource.get_import_id_fields()[0] 

51 self.pk_field = self.resource.fields[pk_field_name] 

52 

53 # If the pk field is missing, all instances in dataset are new 

54 # and cache is empty. 

55 self.all_instances = {} 

56 if self.dataset.dict and self.pk_field.column_name in self.dataset.dict[0]: 

57 ids = [self.pk_field.clean(row) for row in self.dataset.dict] 

58 qs = self.get_queryset().filter(**{ 

59 "%s__in" % self.pk_field.attribute: ids 

60 }) 

61 

62 self.all_instances = { 

63 self.pk_field.get_value(instance): instance 

64 for instance in qs 

65 } 

66 

67 def get_instance(self, row): 

68 if self.all_instances: 

69 return self.all_instances.get(self.pk_field.clean(row)) 

70 return None