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

1from portality.dao import DomainObject 

2from datetime import datetime 

3from copy import deepcopy 

4 

5 

6class PreservationState(DomainObject): 

7 """~~Preservation:Model~~ 

8 Model class for preservation feature""" 

9 __type__ = "preserve" 

10 

11 @property 

12 def status(self): 

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

14 

15 @status.setter 

16 def status(self, value): 

17 self.data["status"] = value 

18 

19 @property 

20 def filename(self): 

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

22 

23 @property 

24 def owner(self): 

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

26 

27 @property 

28 def background_task_id(self): 

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

30 

31 @background_task_id.setter 

32 def background_task_id(self, value): 

33 self.data["background_task_id"] = value 

34 

35 @property 

36 def created_timestamp(self): 

37 if "created_date" not in self.data: 

38 return None 

39 return datetime.strptime(self.data["created_date"], "%Y-%m-%dT%H:%M:%SZ") 

40 

41 @property 

42 def error(self): 

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

44 

45 @property 

46 def error_details(self): 

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

48 

49 @property 

50 def articles_info(self): 

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

52 

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

54 self.data["filename"] = filename 

55 self.data["owner"] = owner 

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

57 self.status = status 

58 

59 def validated(self): 

60 self.status = "validated" 

61 

62 def pending(self): 

63 self.status = "pending" 

64 

65 def uploaded_to_ia(self): 

66 self.status = "uploaded" 

67 

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

69 self.status = "failed" 

70 self.data["error"] = message 

71 if details is not None: 

72 self.data["error_details"] = details 

73 

74 def partial(self): 

75 self.status = "partial" 

76 

77 def successful_articles(self, articles_list): 

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

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

80 

81 def unowned_articles(self, articles_list): 

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

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

84 

85 def no_identifier_articles(self, articles_list): 

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

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

88 

89 def unbagged_articles(self, articles_list): 

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

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

92 

93 def not_found_articles(self, articles_list): 

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

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

96 

97 def no_files_articles(self, articles_list): 

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

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

100 

101 @classmethod 

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

103 q = OwnerFileQuery(owner) 

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

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

106 return rs 

107 

108 

109class OwnerFileQuery(object): 

110 base_query = { 

111 "query": { 

112 "bool": { 

113 "must": [] 

114 } 

115 }, 

116 "sort": [ 

117 {"created_date": "desc"} 

118 ], 

119 "size": 10 

120 } 

121 

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

123 self._query = deepcopy(self.base_query) 

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

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

126 self._query["size"] = size 

127 

128 def query(self): 

129 return self._query