Coverage for portality / models / news.py: 63%

1from portality.dao import DomainObject as DomainObject 

2from copy import deepcopy 

3 

4from portality.lib import dates 

5 

6 

7class News(DomainObject): 

8 """~~News:Model~~""" 

9 __type__ = "news" 

10 

11 @classmethod 

12 def by_remote_id(cls, remote_id): 

13 # ~~->News:Query~~ 

14 q = NewsQuery(remote_id) 

15 es_result = cls.query(q=q.query()) 

16 records = [News(**r.get("_source")) for r in es_result.get("hits", {}).get("hits", [])] 

17 return records 

18 

19 @classmethod 

20 def latest(cls, n): 

21 q = NewsQuery(size=n) 

22 es_result = cls.query(q=q.query()) 

23 records = [News(**r.get("_source")) for r in es_result.get("hits", {}).get("hits", [])] 

24 return records 

25 

26 @property 

27 def remote_id(self): return self.data.get("remote_id") 

28 @remote_id.setter 

29 def remote_id(self, rid): self.data["remote_id"] = rid 

30 

31 @property 

32 def url(self): return self.data.get("url") 

33 @url.setter 

34 def url(self, link): self.data["url"] = link 

35 

36 @property 

37 def title(self): return self.data.get("title") 

38 @title.setter 

39 def title(self, t): self.data["title"] = t 

40 

41 @property 

42 def updated(self): return self.data.get("updated") 

43 @updated.setter 

44 def updated(self, date): self.data["updated"] = date 

45 

46 @property 

47 def published(self): return self.data.get("published") 

48 @published.setter 

49 def published(self, date): self.data["published"] = date 

50 

51 @property 

52 def summary(self): return self.data.get("summary") 

53 @summary.setter 

54 def summary(self, s): self.data["summary"] = s 

55 

56 def published_formatted(self, format="%-d %B %Y"): 

57 try: 

58 dt = dates.parse(self.published) 

59 return dt.strftime(format) 

60 except: 

61 return self.published 

62 

63 

64class NewsQuery(object): 

65 """~~News:Query->Elasticsearch:Technology~~""" 

66 _remote_term = { "term" : { "remote_id.exact" : "<remote id>" } } 

67 

68 def __init__(self, remote_id=None, size=5): 

69 self.remote_id = remote_id 

70 self.size = size 

71 

72 def query(self): 

73 q = {"track_total_hits": True, "query" : {}, "size" : self.size, "sort" : {"published" : {"order" : "desc", "unmapped_type" : "date"}}} 

74 if self.remote_id is not None: 

75 rt = deepcopy(self._remote_term) 

76 rt["term"]["remote_id.exact"] = self.remote_id 

77 q["query"].update(rt) 

78 else: 

79 q["query"]["match_all"] = {} 

80 return q