Coverage for portality / models / preservation.py: 74%

1from portality.dao import DomainObject 

2from copy import deepcopy 

3 

4from portality.lib import dates 

5 

6 

7class PreservationState(DomainObject): 

8 """~~Preservation:Model~~ 

9 Model class for preservation feature""" 

10 __type__ = "preserve" 

11 

12 @property 

13 def status(self): 

14 return self.data.get("status") 

15 

16 @status.setter 

17 def status(self, value): 

18 self.data["status"] = value 

19 

20 @property 

21 def filename(self): 

22 return self.data.get("filename") 

23 

24 @property 

25 def owner(self): 

26 return self.data.get("owner") 

27 

28 @property 

29 def background_task_id(self): 

30 return self.data.get("background_task_id") 

31 

32 @background_task_id.setter 

33 def background_task_id(self, value): 

34 self.data["background_task_id"] = value 

35 

36 @property 

37 def created_timestamp(self): 

38 if "created_date" not in self.data: 

39 return None 

40 return dates.parse(self.data["created_date"]) 

41 

42 @property 

43 def error(self): 

44 return self.data.get("error") 

45 

46 @property 

47 def error_details(self): 

48 return self.data.get("error_details") 

49 

50 @property 

51 def articles_info(self): 

52 return self.data.get("articles_info", {}) 

53 

54 def initiated(self, owner, filename, status="initiated"): 

55 self.data["filename"] = filename 

56 self.data["owner"] = owner 

57 self.data["articles_info"] = {} 

58 self.status = status 

59 

60 def validated(self): 

61 self.status = "validated" 

62 

63 def pending(self): 

64 self.status = "pending" 

65 

66 def uploaded_to_ia(self): 

67 self.status = "uploaded" 

68 

69 def failed(self, message, details=None): 

70 self.status = "failed" 

71 self.data["error"] = message 

72 if details is not None: 

73 self.data["error_details"] = details 

74 

75 def partial(self): 

76 self.status = "partial" 

77 

78 def successful_articles(self, articles_list): 

79 if articles_list is not None and len(articles_list) > 0: 

80 self.data["articles_info"]["successful_articles"] = ", ".join(articles_list) 

81 

82 def unowned_articles(self, articles_list): 

83 if articles_list is not None and len(articles_list) > 0: 

84 self.data["articles_info"]["unowned_articles"] = ", ".join(articles_list) 

85 

86 def no_identifier_articles(self, articles_list): 

87 if articles_list is not None and len(articles_list) > 0: 

88 self.data["articles_info"]["no_identifier_articles"] = ", ".join(articles_list) 

89 

90 def unbagged_articles(self, articles_list): 

91 if articles_list is not None and len(articles_list) > 0: 

92 self.data["articles_info"]["unbagged_articles"] = ", ".join(articles_list) 

93 

94 def not_found_articles(self, articles_list): 

95 if articles_list is not None and len(articles_list) > 0: 

96 self.data["articles_info"]["not_found_articles"] = ", ".join(articles_list) 

97 

98 def no_files_articles(self, articles_list): 

99 if articles_list is not None and len(articles_list) > 0: 

100 self.data["articles_info"]["no_files_articles"] = ", ".join(articles_list) 

101 

102 def uploaded_journals(self, uploaded_journals): 

103 if uploaded_journals is not None and len(uploaded_journals) > 0: 

104 self.data["articles_info"]["uploaded_journals"] = ", ".join(uploaded_journals) 

105 

106 @classmethod 

107 def by_owner(cls, owner, size=10): 

108 q = OwnerFileQuery(owner) 

109 res = cls.query(q=q.query()) 

110 rs = [PreservationState(**r.get("_source")) for r in res.get("hits", {}).get("hits", [])] 

111 return rs 

112 

113 

114class OwnerFileQuery(object): 

115 base_query = { 

116 "query": { 

117 "bool": { 

118 "must": [] 

119 } 

120 }, 

121 "sort": [ 

122 {"created_date": "desc"} 

123 ], 

124 "size": 10 

125 } 

126 

127 def __init__(self, owner, size=10): 

128 self._query = deepcopy(self.base_query) 

129 owner_term = {"match": {"owner.exact": owner}} 

130 self._query["query"]["bool"]["must"].append(owner_term) 

131 self._query["size"] = size 

132 

133 def query(self): 

134 return self._query