Coverage for portality / models / datalog_journal_added.py: 90%

1from __future__ import annotations 

2 

3import elasticsearch 

4 

5from portality import dao 

6from portality.dao import DomainObject, ScrollInitialiseException 

7from portality.lib import coerce 

8from portality.lib.coerce import COERCE_MAP 

9from portality.lib.seamless import SeamlessMixin 

10 

11 

12class DatalogJournalAdded(SeamlessMixin, DomainObject): 

13 __type__ = "datalog_journal_added" 

14 DATE_FMT = '%d-%B-%Y' 

15 

16 __SEAMLESS_STRUCT__ = { 

17 "fields": { 

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

19 "title": {"coerce": "unicode"}, 

20 "issn": {"coerce": "unicode"}, 

21 "date_added": {"coerce": "utcdatetime-datalog"}, 

22 "has_continuations": {"coerce": "bool"}, 

23 "journal_id": {"coerce": "unicode"}, 

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

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

26 'es_type': {'coerce': 'unicode'}, 

27 }, 

28 } 

29 

30 __SEAMLESS_COERCE__ = { 

31 **COERCE_MAP, 

32 "utcdatetime-datalog": coerce.date_str(in_format=DATE_FMT), 

33 } 

34 

35 def __init__(self, **kwargs): 

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

37 

38 @property 

39 def data(self): 

40 return self.__seamless__.data 

41 

42 @property 

43 def title(self): 

44 return self.__seamless__.get_single('title') 

45 

46 @title.setter 

47 def title(self, val): 

48 self.__seamless__.set_single('title', val) 

49 

50 @property 

51 def issn(self): 

52 return self.__seamless__.get_single("issn") 

53 

54 @issn.setter 

55 def issn(self, val): 

56 self.__seamless__.set_single('issn', val) 

57 

58 @property 

59 def date_added(self): 

60 return self.__seamless__.get_single("date_added") 

61 

62 @date_added.setter 

63 def date_added(self, val): 

64 self.__seamless__.set_with_struct('date_added', val) 

65 

66 @property 

67 def date_added_str(self): 

68 return self.date_added.strftime(self.DATE_FMT) 

69 

70 @property 

71 def has_continuations(self): 

72 return self.__seamless__.get_single("has_continuations") 

73 

74 @has_continuations.setter 

75 def has_continuations(self, val): 

76 self.__seamless__.set_single('has_continuations', val) 

77 

78 @property 

79 def journal_id(self): 

80 return self.__seamless__.get_single("journal_id") 

81 

82 @journal_id.setter 

83 def journal_id(self, val): 

84 self.__seamless__.set_single('journal_id', val) 

85 

86 

87class LastDatalogJournalAddedQuery: 

88 

89 def query(self): 

90 return { 

91 "size": 1, 

92 "sort": [ 

93 { 

94 "date_added": { 

95 "order": "desc" 

96 } 

97 } 

98 ], 

99 "query": { 

100 "match_all": {} 

101 } 

102 } 

103 

104 

105def find_last_datalog(): 

106 try: 

107 record = next(DatalogJournalAdded.iterate(LastDatalogJournalAddedQuery().query()), None) 

108 except (elasticsearch.exceptions.NotFoundError, ScrollInitialiseException): 

109 record = None 

110 return record 

111 

112 

113class DateAddedDescQuery: 

114 

115 def query(self): 

116 return { 

117 'sort': [ 

118 {'date_added': {'order': 'desc'}} 

119 ] 

120 } 

121 

122 

123class IssnDateMatchQuery: 

124 def __init__(self, issn, date_added): 

125 self.issn = issn 

126 self.date_added = date_added 

127 

128 def query(self): 

129 return { 

130 "query": { 

131 "bool": { 

132 "filter": [ 

133 {"term": {"issn.keyword": self.issn}}, 

134 {"term": {"date_added": self.date_added}} 

135 ] 

136 } 

137 }, 

138 } 

139 

140 

141def is_issn_exists(issn, date_added): 

142 return dao.is_exist(IssnDateMatchQuery(issn, date_added).query(), DatalogJournalAdded.index_name())