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 

11 

12blueprint = Blueprint('atom', __name__) 

13 

14 

15@blueprint.route('/feed') 

16@plausible.pa_event(app.config.get('GA_CATEGORY_ATOM', 'Atom'), 

17 action=app.config.get('GA_ACTION_ACTION', 'Feed Request')) 

18def feed(): 

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

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

21 # is testable) 

22 f = get_feed(request.base_url) 

23 

24 # serialise and respond with the atom xml 

25 resp = make_response(f.serialise()) 

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

27 return resp 

28 

29 

30def get_feed(base_url=None): 

31 """ 

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

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

34 be serialised and delivered by the web layer 

35 

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

37 :return: AtomFeed object 

38 """ 

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

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

41 from_date = (datetime.now() - timedelta(0, max_age)).strftime("%Y-%m-%dT%H:%M:%SZ") 

42 

43 dao = models.AtomRecord() 

44 records = dao.list_records(from_date, max_size) 

45 

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

47 url = base_url 

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

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

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

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

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

53 

54 xwalk = AtomCrosswalk() 

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

56 

57 for record in records: 

58 entry = xwalk.crosswalk(record) 

59 f.add_entry(entry) 

60 

61 return f 

62 

63 

64class AtomFeed(object): 

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

66 ATOM = "{%s}" % ATOM_NAMESPACE 

67 NSMAP = {None: ATOM_NAMESPACE} 

68 

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

70 self.title = title 

71 self.url = url 

72 self.generator = generator 

73 self.icon = icon 

74 self.logo = logo 

75 self.link = link 

76 self.rights = rights 

77 self.last_updated = None 

78 self.entries = {} 

79 

80 def add_entry(self, entry): 

81 # update the "last_updated" property if necessary 

82 lu = entry.get("updated") 

83 dr = datetime.strptime(lu, "%Y-%m-%dT%H:%M:%SZ") 

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

85 self.last_updated = dr 

86 

87 # record the entries by date 

88 if lu in self.entries: 

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

90 else: 

91 self.entries[lu] = [entry] 

92 

93 def serialise(self): 

94 if self.last_updated is None: 

95 self.last_updated = datetime.now() 

96 

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

98 

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

100 title.text = self.title 

101 

102 if self.generator is not None: 

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

104 generator.text = self.generator 

105 

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

107 icon.text = self.icon 

108 

109 if self.logo is not None: 

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

111 logo.text = self.logo 

112 

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

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

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

116 

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

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

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

120 

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

122 rights.text = self.rights 

123 

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

125 dr = datetime.strftime(self.last_updated, "%Y-%m-%dT%H:%M:%SZ") 

126 updated.text = dr 

127 

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

129 entry_dates.sort(reverse=True) 

130 for ed in entry_dates: 

131 es = self.entries.get(ed) 

132 for e in es: 

133 self._serialise_entry(feed, e) 

134 

135 tree = etree.ElementTree(feed) 

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

137 

138 def _serialise_entry(self, feed, e): 

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

140 

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

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

143 name.text = e['author'] 

144 

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

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

147 c.set("term", cat) 

148 

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

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

151 

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

153 id.text = e['id'] 

154 

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

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

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

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

159 

160 if "related" in e: 

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

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

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

164 

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

166 rights.text = e['rights'] 

167 

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

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

170 summary.text = e['summary'] 

171 

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

173 title.text = e['title'] 

174 

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

176 updated.text = e['updated']