Coverage for portality / view / atom.py: 95%

1from flask import Blueprint, request, make_response 

2 

3from portality import models as models 

4from portality.core import app 

5from portality.crosswalks.atom import AtomCrosswalk 

6 

7from lxml import etree 

8from datetime import datetime, timedelta 

9 

10from portality.lib import plausible, dates 

11from portality.lib.dates import FMT_DATETIME_STD 

12 

13blueprint = Blueprint('atom', __name__) 

14 

15 

16@blueprint.route('/feed') 

17@plausible.pa_event(app.config.get('ANALYTICS_CATEGORY_ATOM', 'Atom'), 

18 action=app.config.get('ANALYTICS_ACTION_ACTION', 'Feed Request')) 

19def feed(): 

20 # get the feed for this base_url (which is just used to set the metadata of 

21 # the feed, but we want to do this outside of a request context so it 

22 # is testable) 

23 f = get_feed(request.base_url) 

24 

25 # serialise and respond with the atom xml 

26 resp = make_response(f.serialise()) 

27 resp.mimetype = "application/atom+xml" 

28 return resp 

29 

30 

31def get_feed(base_url=None): 

32 """ 

33 Main method for generating the feed. Gets all of the settings 

34 out of config and returns the feed object, which can then 

35 be serialised and delivered by the web layer 

36 

37 :param base_url: The base url to include in the feed metadata 

38 :return: AtomFeed object 

39 """ 

40 max_size = app.config.get("MAX_FEED_ENTRIES", 20) 

41 max_age = app.config.get("MAX_FEED_ENTRY_AGE", 2592000) 

42 from_date = (dates.now() - timedelta(0, max_age)).strftime(FMT_DATETIME_STD) 

43 

44 dao = models.AtomRecord() 

45 records = dao.list_records(from_date, max_size) 

46 

47 title = app.config.get("FEED_TITLE", "untitled") 

48 url = base_url 

49 generator = app.config.get('FEED_GENERATOR', "") 

50 icon = app.config.get("FEED_LOGO", "") 

51 logo = app.config.get("FEED_LOGO", "") 

52 link = app.config.get('BASE_URL', "") 

53 rights = app.config.get('FEED_LICENCE', "") 

54 

55 xwalk = AtomCrosswalk() 

56 f = AtomFeed(title, url, generator, icon, logo, link, rights) 

57 

58 for record in records: 

59 entry = xwalk.crosswalk(record) 

60 f.add_entry(entry) 

61 

62 return f 

63 

64 

65class AtomFeed(object): 

66 ATOM_NAMESPACE = "http://www.w3.org/2005/Atom" 

67 ATOM = "{%s}" % ATOM_NAMESPACE 

68 NSMAP = {None: ATOM_NAMESPACE} 

69 

70 def __init__(self, title, url, generator, icon, logo, link, rights): 

71 self.title = title 

72 self.url = url 

73 self.generator = generator 

74 self.icon = icon 

75 self.logo = logo 

76 self.link = link 

77 self.rights = rights 

78 self.last_updated = None 

79 self.entries = {} 

80 

81 def add_entry(self, entry): 

82 # update the "last_updated" property if necessary 

83 lu = entry.get("updated") 

84 dr = dates.parse(lu) 

85 if self.last_updated is None or dr > self.last_updated: 

86 self.last_updated = dr 

87 

88 # record the entries by date 

89 if lu in self.entries: 

90 self.entries[lu].append(entry) 

91 else: 

92 self.entries[lu] = [entry] 

93 

94 def serialise(self): 

95 if self.last_updated is None: 

96 self.last_updated = dates.now() 

97 

98 feed = etree.Element(self.ATOM + "feed", nsmap=self.NSMAP) 

99 

100 title = etree.SubElement(feed, self.ATOM + "title") 

101 title.text = self.title 

102 

103 if self.generator is not None: 

104 generator = etree.SubElement(feed, self.ATOM + "generator") 

105 generator.text = self.generator 

106 

107 icon = etree.SubElement(feed, self.ATOM + "icon") 

108 icon.text = self.icon 

109 

110 if self.logo is not None: 

111 logo = etree.SubElement(feed, self.ATOM + "logo") 

112 logo.text = self.logo 

113 

114 self_link = etree.SubElement(feed, self.ATOM + "link") 

115 self_link.set("rel", "self") 

116 self_link.set("href", self.url) 

117 

118 link = etree.SubElement(feed, self.ATOM + "link") 

119 link.set("rel", "related") 

120 link.set("href", self.link) 

121 

122 rights = etree.SubElement(feed, self.ATOM + "rights") 

123 rights.text = self.rights 

124 

125 updated = etree.SubElement(feed, self.ATOM + "updated") 

126 dr = datetime.strftime(self.last_updated, FMT_DATETIME_STD) 

127 updated.text = dr 

128 

129 entry_dates = list(self.entries.keys()) 

130 entry_dates.sort(reverse=True) 

131 for ed in entry_dates: 

132 es = self.entries.get(ed) 

133 for e in es: 

134 self._serialise_entry(feed, e) 

135 

136 tree = etree.ElementTree(feed) 

137 return etree.tostring(tree, pretty_print=True, xml_declaration=True, encoding="utf-8") 

138 

139 def _serialise_entry(self, feed, e): 

140 entry = etree.SubElement(feed, self.ATOM + "entry") 

141 

142 author = etree.SubElement(entry, self.ATOM + "author") 

143 name = etree.SubElement(author, self.ATOM + "name") 

144 name.text = e['author'] 

145 

146 for cat in e.get("categories", []): 

147 c = etree.SubElement(entry, self.ATOM + "category") 

148 c.set("term", cat) 

149 

150 cont = etree.SubElement(entry, self.ATOM + "content") 

151 cont.set("src", e['content_src']) 

152 

153 id = etree.SubElement(entry, self.ATOM + "id") 

154 id.text = e['id'] 

155 

156 # this is not strictly necessary, as we have an atom:content element, but it can't harm 

157 alt = etree.SubElement(entry, self.ATOM + "link") 

158 alt.set("rel", "alternate") 

159 alt.set("href", e['alternate']) 

160 

161 if "related" in e: 

162 rel = etree.SubElement(entry, self.ATOM + "link") 

163 rel.set("rel", "related") 

164 rel.set("href", e['related']) 

165 

166 rights = etree.SubElement(entry, self.ATOM + "rights") 

167 rights.text = e['rights'] 

168 

169 summary = etree.SubElement(entry, self.ATOM + "summary") 

170 summary.set("type", "text") 

171 summary.text = e['summary'] 

172 

173 title = etree.SubElement(entry, self.ATOM + "title") 

174 title.text = e['title'] 

175 

176 updated = etree.SubElement(entry, self.ATOM + "updated") 

177 updated.text = e['updated']