Coverage for portality / models / v2 / application.py: 80%
217 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-05 00:09 +0100
1from copy import deepcopy
3from portality import constants
4from portality.core import app
5from portality.lib import es_data_mapping, coerce, dates
6from portality.models.v2 import shared_structs
7from portality.models.v2.journal import JournalLikeObject, Journal
8from portality.lib.coerce import COERCE_MAP
9from portality.dao import DomainObject
10from portality.bll import DOAJ
14APPLICATION_STRUCT = {
15 "objects": [
16 "admin", "index"
17 ],
19 "structs": {
20 "admin": {
21 "fields": {
22 "current_journal": {"coerce": "unicode"},
23 "related_journal": {"coerce": "unicode"},
24 "application_status": {"coerce": "unicode"},
25 "date_rejected": {"coerce": "utcdatetime"},
26 "application_type": {
27 "coerce": "unicode",
28 "allowed_values": [
29 constants.APPLICATION_TYPE_NEW_APPLICATION,
30 constants.APPLICATION_TYPE_UPDATE_REQUEST
31 ]
32 }
33 }
34 },
35 "index": {
36 "fields": {
37 "application_type": {"coerce": "unicode"}
38 }
39 }
40 }
41}
43# ~~Application:Model~~
44class Application(JournalLikeObject):
45 __type__ = "application"
47 __SEAMLESS_STRUCT__ = [
48 shared_structs.JOURNAL_BIBJSON,
49 shared_structs.SHARED_JOURNAL_LIKE,
50 APPLICATION_STRUCT
51 ]
53 __SEAMLESS_COERCE__ = COERCE_MAP
55 def __init__(self, **kwargs):
56 # FIXME: hack, to deal with ES integration layer being improperly abstracted
57 if "_source" in kwargs:
58 kwargs = kwargs["_source"]
59 super(Application, self).__init__(raw=kwargs)
60 if self.application_type is None:
61 if self.current_journal:
62 self.set_is_update_request(True)
63 else:
64 self.set_is_update_request(False)
66 @classmethod
67 def get_by_owner(cls, owner):
68 q = SuggestionQuery(owner=owner)
69 result = cls.query(q=q.query())
70 records = [cls(**r.get("_source")) for r in result.get("hits", {}).get("hits", [])]
71 return records
73 @classmethod
74 def delete_selected(cls, email=None, statuses=None):
75 q = SuggestionQuery(email=email, statuses=statuses)
76 r = cls.delete_by_query(q.query())
78 @classmethod
79 def list_by_status(cls, status):
80 q = StatusQuery(status)
81 return cls.iterate(q=q.query())
83 @classmethod
84 def find_latest_by_current_journal(cls, journal_id):
85 q = CurrentJournalQuery(journal_id)
86 results = cls.q2obj(q=q.query())
87 if len(results) > 0:
88 return results[0]
89 return None
91 @classmethod
92 def find_all_by_related_journal(cls, journal_id):
93 q = RelatedJournalQuery(journal_id, size=1000)
94 return cls.q2obj(q=q.query())
96 @classmethod
97 def assignment_to_editor_groups(cls, egs):
98 q = AssignedEditorGroupsQuery([eg.name for eg in egs])
99 res = cls.query(q.query())
100 buckets = res.get("aggregations", {}).get("editor_groups", {}).get("buckets", [])
101 assignments = {}
102 for b in buckets:
103 assignments[b.get("key")] = b.get("doc_count")
104 return assignments
106 def mappings(self):
107 return es_data_mapping.create_mapping(self.__seamless_struct__.raw, MAPPING_OPTS)
109 @property
110 def application_type(self):
111 return self.__seamless__.get_single("admin.application_type")
113 @application_type.setter
114 def application_type(self, val):
115 self.__seamless__.set_with_struct("admin.application_type", val)
117 def set_is_update_request(self, val):
118 application_type = constants.APPLICATION_TYPE_UPDATE_REQUEST if val is True else constants.APPLICATION_TYPE_NEW_APPLICATION
119 self.application_type = application_type
121 @property
122 def current_journal(self):
123 return self.__seamless__.get_single("admin.current_journal")
125 def set_current_journal(self, journal_id):
126 # anything that has a current journal is, by definition, an update request
127 self.set_is_update_request(True)
128 self.__seamless__.set_with_struct("admin.current_journal", journal_id)
130 def remove_current_journal(self):
131 self.__seamless__.delete("admin.current_journal")
133 @property
134 def related_journal(self):
135 return self.__seamless__.get_single("admin.related_journal")
137 def set_related_journal(self, journal_id):
138 self.__seamless__.set_with_struct("admin.related_journal", journal_id)
140 def remove_related_journal(self):
141 self.__seamless__.delete("admin.related_journal")
143 @property
144 def related_journal_object(self):
145 if self.related_journal:
146 return Journal.pull(self.related_journal)
147 return None
149 @property
150 def application_status(self):
151 return self.__seamless__.get_single("admin.application_status")
153 def set_application_status(self, val):
154 self.__seamless__.set_with_struct("admin.application_status", val)
156 @property
157 def date_rejected(self):
158 return self.__seamless__.get_single("admin.date_rejected")
160 @property
161 def date_rejected_timestamp(self):
162 return self.__seamless__.get_single("admin.date_rejected", coerce=coerce.to_datestamp())
164 @date_rejected.setter
165 def date_rejected(self, value):
166 self.__seamless__.set_with_struct("admin.date_rejected", value)
168 def _sync_owner_to_journal(self):
169 if self.current_journal is None:
170 return
171 from portality.models.v2.journal import Journal
172 cj = Journal.pull(self.current_journal)
173 if cj is not None and cj.owner != self.owner:
174 cj.set_owner(self.owner)
175 cj.save(sync_owner=False)
177 def _generate_index(self):
178 super(Application, self)._generate_index()
180 # index_record_type = None
181 # if self.application_type == constants.APPLICATION_TYPE_NEW_APPLICATION:
182 # if self.application_status in [constants.APPLICATION_STATUS_REJECTED, constants.APPLICATION_STATUS_ACCEPTED]:
183 # index_record_type = constants.INDEX_RECORD_TYPE_NEW_APPLICATION_FINISHED
184 # else:
185 # index_record_type = constants.INDEX_RECORD_TYPE_NEW_APPLICATION_UNFINISHED
186 # elif self.application_type == constants.APPLICATION_TYPE_UPDATE_REQUEST:
187 # if self.application_status in [constants.APPLICATION_STATUS_REJECTED, constants.APPLICATION_STATUS_ACCEPTED]:
188 # index_record_type = constants.INDEX_RECORD_TYPE_UPDATE_REQUEST_FINISHED
189 # else:
190 # index_record_type = constants.INDEX_RECORD_TYPE_UPDATE_REQUEST_UNFINISHED
191 # if index_record_type is not None:
192 # self.__seamless__.set_with_struct("index.application_type", index_record_type)
194 # FIXME: Temporary partial reversion of an indexing change (this index.application_type data is still being used
195 # in applications search)
196 if self.current_journal is not None:
197 self.__seamless__.set_with_struct("index.application_type", "update request")
198 elif self.application_status in [constants.APPLICATION_STATUS_ACCEPTED, constants.APPLICATION_STATUS_REJECTED]:
199 self.__seamless__.set_with_struct("index.application_type", "finished application/update")
200 else:
201 self.__seamless__.set_with_struct("index.application_type", "new application")
203 def prep(self, is_update=True):
204 self._generate_index()
205 if is_update:
206 self.set_last_updated()
208 def save(self, sync_owner=True, **kwargs):
209 if self.id is None:
210 self.set_id(self.makeid())
212 if self.application_type == constants.APPLICATION_TYPE_UPDATE_REQUEST:
213 # ~~-> Concurrency_Prevention:Service ~~
214 cs = DOAJ.applicationService()
215 cs.prevent_concurrent_ur_submission(self, record_if_not_concurrent=True)
217 self.prep()
218 self.verify_against_struct()
220 if sync_owner:
221 self._sync_owner_to_journal()
222 return super(Application, self).save(**kwargs)
224 #########################################################
225 ## DEPRECATED METHODS
227 @property
228 def suggested_on(self):
229 return self.date_applied
231 @suggested_on.setter
232 def suggested_on(self, val):
233 self.date_applied = val
235 @property
236 def suggester(self):
237 owner = self.owner
238 if owner is None:
239 return None
240 from portality.models import Account
241 oacc = Account.pull(owner)
242 if oacc is None:
243 return None
244 return {
245 "name": owner,
246 "email": oacc.email
247 }
250class DraftApplication(Application):
251 __type__ = "draft_application"
253 __SEAMLESS_APPLY_STRUCT_ON_INIT__ = False
254 __SEAMLESS_CHECK_REQUIRED_ON_INIT__ = False
255 __SEAMLESS_SILENT_PRUNE__ = True
258class AllPublisherApplications(DomainObject):
259 __type__ = "draft_application,application"
262MAPPING_OPTS = {
263 "dynamic": None,
264 "coerces": Journal.add_mapping_extensions(app.config["DATAOBJ_TO_MAPPING_DEFAULTS"]),
265 "exceptions": {**app.config["ADMIN_NOTES_SEARCH_MAPPING"], **app.config["JOURNAL_EXCEPTION_MAPPING"]},
266 "additional_mappings": app.config["ADMIN_NOTES_INDEX_ONLY_FIELDS"]
267}
270class SuggestionQuery(object):
271 _base_query = {"track_total_hits": True, "query": {"bool": {"must": []}}}
272 _email_term = {"term": {"admin.applicant.email.exact": "<email address>"}}
273 _status_terms = {"terms": {"admin.application_status.exact": ["<list of statuses>"]}}
274 _owner_term = {"term": {"admin.owner.exact": "<the owner id>"}}
276 def __init__(self, email=None, statuses=None, owner=None):
277 self.email = email
278 self.owner = owner
279 if statuses:
280 self.statuses = statuses
281 else:
282 self.statuses = []
284 def query(self):
285 q = deepcopy(self._base_query)
286 if self.email:
287 et = deepcopy(self._email_term)
288 et["term"]["admin.applicant.email.exact"] = self.email
289 q["query"]["bool"]["must"].append(et)
290 if self.statuses and len(self.statuses) > 0:
291 st = deepcopy(self._status_terms)
292 st["terms"]["admin.application_status.exact"] = self.statuses
293 q["query"]["bool"]["must"].append(st)
295 if self.owner is not None:
296 ot = deepcopy(self._owner_term)
297 ot["term"]["admin.owner.exact"] = self.owner
298 q["query"]["bool"]["must"].append(ot)
299 return q
302class OwnerStatusQuery(object):
303 base_query = {
304 "track_total_hits": True,
305 "query": {
306 "bool": {
307 "must": []
308 }
309 },
310 "sort": [
311 {"created_date": "desc"}
312 ],
313 "size": 10
314 }
316 def __init__(self, owner, statuses, size=10):
317 self._query = deepcopy(self.base_query)
318 owner_term = {"match": {"owner": owner}}
319 self._query["query"]["bool"]["must"].append(owner_term)
320 status_term = {"terms": {"admin.application_status.exact": statuses}}
321 self._query["query"]["bool"]["must"].append(status_term)
322 self._query["size"] = size
324 def query(self):
325 return self._query
328class StatusQuery(object):
330 def __init__(self, statuses):
331 if not isinstance(statuses, list):
332 statuses = [statuses]
333 self.statuses = statuses
335 def query(self):
336 return {
337 "track_total_hits": True,
338 "query": {
339 "bool": {
340 "must": [
341 {"terms": {"admin.application_status.exact": self.statuses}}
342 ]
343 }
344 }
345 }
348class CurrentJournalQuery(object):
350 def __init__(self, journal_id, size=1):
351 self.journal_id = journal_id
352 self.size = size
354 def query(self):
355 return {
356 "track_total_hits": True,
357 "query": {
358 "bool": {
359 "must": [
360 {"term": {"admin.current_journal.exact": self.journal_id}}
361 ]
362 }
363 },
364 "sort": [
365 {"admin.date_applied": {"order": "desc"}}
366 ],
367 "size": self.size
368 }
371class RelatedJournalQuery(object):
373 def __init__(self, journal_id, size=1):
374 self.journal_id = journal_id
375 self.size = size
377 def query(self):
378 return {
379 "track_total_hits": True,
380 "query": {
381 "bool": {
382 "must": [
383 {"term": {"admin.related_journal.exact": self.journal_id}}
384 ]
385 }
386 },
387 "sort": [
388 {"admin.date_applied": {"order": "asc"}}
389 ],
390 "size": self.size
391 }
394class AssignedEditorGroupsQuery(object):
395 """
396 ~~->$AssignedEditorGroups:Query~~
397 ~~^->Elasticsearch:Technology~~
398 """
400 def __init__(self, editor_groups, application_type_name="new application"):
401 self.editor_groups = editor_groups
402 self.application_type_name = application_type_name
404 def query(self):
405 return {
406 "query": {
407 "bool": {
408 "must": [
409 {"terms": {"admin.editor_group.exact": self.editor_groups}},
410 {"term": {"index.application_type.exact": self.application_type_name}}
411 ]
412 }
413 },
414 "size": 0,
415 "aggs": {
416 "editor_groups": {
417 "terms": {
418 "field": "admin.editor_group.exact",
419 "size": len(self.editor_groups) or 1,
420 "order": {"_term": "asc"}
421 }
422 }
423 }
424 }