Coverage for apis_core/generic/importers.py: 0%

1import json 

2import logging 

3import urllib 

4from functools import cache 

5 

6from django.core.exceptions import ImproperlyConfigured 

7 

8from apis_core.utils.normalize import clean_uri 

9from apis_core.utils.rdf import get_definition_and_attributes_from_uri 

10 

11logger = logging.getLogger(__name__) 

12 

13 

14@cache 

15class GenericModelImporter: 

16 """ 

17 A generic importer class 

18 It provides the standard methods for importing data from 

19 an URI and creating a model instance of it. 

20 By default it fetches a resource, first tries to parse it using 

21 our rdf parser, if that fails tries to parse it using json and 

22 then extracts the fields whose keys match the model field names. 

23 Projects can inherit from this class and override the default 

24 methods or simple write their own from scratch. 

25 """ 

26 

27 model = None 

28 import_uri = None 

29 

30 def __init__(self, uri, model): 

31 self.model = model 

32 self.import_uri = self.clean_uri(uri) 

33 

34 @property 

35 def get_uri(self): 

36 return self.import_uri 

37 

38 def clean_uri(self, uri): 

39 return clean_uri(uri) 

40 

41 @cache 

42 def request(self, uri): 

43 # we first try to use the RDF parser 

44 try: 

45 defn, data = get_definition_and_attributes_from_uri(uri, self.model) 

46 return data 

47 except Exception as e: 

48 logging.debug(e) 

49 # if everything else fails, try parsing json 

50 # if even that does not help, return an empty dict 

51 try: 

52 return json.loads(urllib.request.urlopen(uri).read()) 

53 except Exception as e: 

54 logging.debug(e) 

55 return {} 

56 

57 def mangle_data(self, data): 

58 return data 

59 

60 def get_data(self, drop_unknown_fields=True): 

61 """ 

62 fetch the data using the `request` method and 

63 mangle the data using the `mangle_data` method. 

64 If the `drop_unknown_fields` argument is true, 

65 remove all fields from the data dict that do not 

66 have an equivalent field in the model. 

67 """ 

68 data = self.request(self.import_uri) 

69 data = self.mangle_data(data) 

70 if drop_unknown_fields: 

71 # we are dropping all fields that are not part of the model 

72 modelfields = [field.name for field in self.model._meta.fields] 

73 data = {key: data[key] for key in data if key in modelfields} 

74 if not data: 

75 raise ImproperlyConfigured( 

76 f"Could not import {self.import_uri}. Data fetched was: {data}" 

77 ) 

78 return data 

79 

80 def import_into_instance(self, instance, fields="__all__"): 

81 data = self.get_data() 

82 if fields == "__all__": 

83 fields = data.keys() 

84 for field in fields: 

85 if hasattr(instance, field) and field in data.keys(): 

86 setattr(instance, field, data[field]) 

87 instance.save() 

88 

89 def create_instance(self): 

90 return self.model.objects.create(**self.get_data(drop_unknown_fields=True))