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

1from portality.dao import DomainObject as DomainObject 

2from copy import deepcopy 

3from datetime import datetime 

4 

5 

6class News(DomainObject): 

7 """~~News:Model~~""" 

8 __type__ = "news" 

9 

10 @classmethod 

11 def by_remote_id(cls, remote_id): 

12 # ~~->News:Query~~ 

13 q = NewsQuery(remote_id) 

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

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

16 return records 

17 

18 @classmethod 

19 def latest(cls, n): 

20 q = NewsQuery(size=n) 

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

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

23 return records 

24 

25 @property 

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

27 @remote_id.setter 

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

29 

30 @property 

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

32 @url.setter 

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

34 

35 @property 

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

37 @title.setter 

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

39 

40 @property 

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

42 @updated.setter 

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

44 

45 @property 

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

47 @published.setter 

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

49 

50 @property 

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

52 @summary.setter 

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

54 

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

56 try: 

57 dt = datetime.strptime(self.published, "%Y-%m-%dT%H:%M:%SZ") 

58 return dt.strftime(format) 

59 except: 

60 return self.published 

61 

62 

63class NewsQuery(object): 

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

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

66 

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

68 self.remote_id = remote_id 

69 self.size = size 

70 

71 def query(self): 

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

73 if self.remote_id is not None: 

74 rt = deepcopy(self._remote_term) 

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

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

77 else: 

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

79 return q