Coverage for portality / models / ris_export.py: 100%

1from io import BytesIO 

2 

3from portality.core import app 

4from portality.dao import DomainObject 

5from portality.lib import es_data_mapping 

6from portality.lib.coerce import COERCE_MAP 

7from portality.lib.ris import RisEntry 

8from portality.lib.seamless import SeamlessMixin 

9from portality.lib import dates 

10 

11STRUCT = { 

12 "fields": { 

13 "id": {"coerce": "unicode"}, 

14 "created_date": {"coerce": "utcdatetime"}, 

15 "last_updated": {"coerce": "utcdatetime"}, 

16 "es_type": {"coerce": "unicode"}, 

17 "ris": {"coerce": "unicode"}, 

18 } 

19} 

20 

21MAPPING_OPTS = { 

22 "dynamic": None, 

23 "coerces": app.config["DATAOBJ_TO_MAPPING_DEFAULTS"] 

24} 

25 

26class RISExport(SeamlessMixin, DomainObject): 

27 __type__ = "ris_export" 

28 

29 __SEAMLESS_STRUCT__ = STRUCT 

30 __SEAMLESS_COERCE__ = COERCE_MAP 

31 

32 def __init__(self, **kwargs): 

33 # FIXME: hack, to deal with ES integration layer being improperly abstracted 

34 if "_source" in kwargs: 

35 kwargs = kwargs["_source"] 

36 super(RISExport, self).__init__(raw=kwargs) 

37 

38 def mappings(self): 

39 return es_data_mapping.create_mapping(self.__seamless_struct__.raw, MAPPING_OPTS) 

40 

41 @property 

42 def data(self): 

43 return self.__seamless__.data 

44 

45 @property 

46 def ris_raw(self): 

47 return self.__seamless__.get_single("ris") 

48 

49 @ris_raw.setter 

50 def ris_raw(self, val): 

51 if isinstance(val, RisEntry): 

52 val = val.to_text() 

53 self.__seamless__.set_with_struct("ris", val) 

54 

55 @property 

56 def byte_stream(self): 

57 raw = self.ris_raw 

58 raw = raw.encode('utf-8', errors='ignore') 

59 bs = BytesIO() 

60 bs.write(raw) 

61 bs.seek(0) 

62 return bs