Coverage for portality / models / editors.py: 89%

1from portality.dao import DomainObject, ScrollInitialiseException 

2from portality.models import Account 

3 

4 

5class EditorGroup(DomainObject): 

6 """ 

7 ~~EditorGroup:Model~~ 

8 """ 

9 __type__ = "editor_group" 

10 

11 @classmethod 

12 def group_exists_by_name(cls, name): 

13 q = EditorGroupQuery(name) 

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

15 ids = [hit.get("_source", {}).get("id") for hit in res.get("hits", {}).get("hits", []) if "_source" in hit] 

16 if len(ids) == 0: 

17 return None 

18 if len(ids) > 0: 

19 return ids[0] 

20 

21 @classmethod 

22 def group_by_name(cls, name): 

23 id = cls.group_exists_by_name(name) 

24 if id is not None: 

25 return cls.pull(id) 

26 return None 

27 

28 @classmethod 

29 def _groups_by_x(cls, **kwargs): 

30 """ Generalised editor groups by maned / editor / associate """ 

31 # ~~-> EditorGroupMember:Query~~ 

32 q = EditorGroupMemberQuery(**kwargs) 

33 try: 

34 return [g for g in cls.iterate(q.query(), page_size=100)] 

35 except ScrollInitialiseException: 

36 # If there's no mapping for editor groups, the account definitely has no assignments. 

37 return [] 

38 

39 @classmethod 

40 def groups_by_maned(cls, maned): 

41 return cls._groups_by_x(maned=maned) 

42 

43 @classmethod 

44 def groups_by_editor(cls, editor): 

45 return cls._groups_by_x(editor=editor) 

46 

47 @classmethod 

48 def groups_by_associate(cls, associate): 

49 return cls._groups_by_x(associate=associate) 

50 

51 @property 

52 def name(self): 

53 return self.data.get("name") 

54 

55 def set_name(self, val): 

56 self.data["name"] = val 

57 

58 @property 

59 def maned(self): 

60 return self.data.get("maned") 

61 

62 def set_maned(self, val): 

63 self.data["maned"] = val 

64 

65 def get_maned_account(self): 

66 return Account.pull(self.maned) 

67 

68 @property 

69 def editor(self): 

70 return self.data.get("editor") 

71 

72 def set_editor(self, val): 

73 self.data["editor"] = val 

74 

75 def get_editor_account(self): 

76 return Account.pull(self.editor) 

77 

78 @property 

79 def associates(self): 

80 return self.data.get("associates", []) 

81 

82 def set_associates(self, val): 

83 if not isinstance(val, list): 

84 val = [val] 

85 self.data["associates"] = val 

86 

87 def add_associate(self, val): 

88 if "associates" not in self.data: 

89 self.data["associates"] = [] 

90 self.data["associates"].append(val) 

91 

92 def get_associate_accounts(self): 

93 accs = [] 

94 for a in self.associates: 

95 acc = Account.pull(a) 

96 accs.append(acc) 

97 return accs 

98 

99 def is_member(self, account_name): 

100 """ Determine if an account is a member of this Editor Group """ 

101 all_eds = self.associates + [self.editor] 

102 return account_name in all_eds 

103 

104 

105class EditorGroupQuery(object): 

106 def __init__(self, name): 

107 self.name = name 

108 

109 def query(self): 

110 q = { 

111 "track_total_hits" : True, 

112 "query": {"term": {"name.exact": self.name}} 

113 } 

114 return q 

115 

116 

117class EditorGroupMemberQuery(object): 

118 """ 

119 ~~EditorGroupMember:Query->Elasticsearch:Technology~~ 

120 """ 

121 def __init__(self, editor=None, associate=None, maned=None): 

122 self.editor = editor 

123 self.associate = associate 

124 self.maned = maned 

125 

126 def query(self): 

127 q = { 

128 "track_total_hits": True, 

129 "query": {"bool": {"should": []}}, 

130 "sort": {"name.exact": {"order" : "asc"}} 

131 } 

132 if self.editor is not None: 

133 et = {"term": {"editor.exact": self.editor}} 

134 q["query"]["bool"]["should"].append(et) 

135 if self.associate is not None: 

136 at = {"term": {"associates.exact": self.associate}} 

137 q["query"]["bool"]["should"].append(at) 

138 if self.maned is not None: 

139 mt = {"term" : {"maned.exact" : self.maned}} 

140 q["query"]["bool"]["should"].append(mt) 

141 return q