AbstractAtomFeedView   A
last analyzed

Complexity

Total Complexity 4

Size/Duplication

Total Lines 30
Duplicated Lines 0 %

Importance

Changes 1
Bugs 0 Features 0
Metric Value
c 1
b 0
f 0
dl 0
loc 30
rs 10
wmc 4

4 Methods

Rating   Name   Duplication   Size   Complexity  
A archive_current_feed() 0 5 1
A get_atom_feed_entry() 0 6 1
A get_atom_feed() 0 6 1
A get_current_atom_feed() 0 6 1
1
# -*- coding: utf-8 -*-
2
from feedgen.feed import FeedGenerator
3
import abc
4
5
6
class AbstractAtomFeedView:
7
8
    __metaclass__ = abc.ABCMeta
9
10
    @abc.abstractmethod
11
    def get_current_atom_feed(self):
12
        """
13
        Retrieve the current atom feed
14
        :return: JSON representation of the feed
15
        """
16
17
    @abc.abstractmethod
18
    def get_atom_feed(self):
19
        """
20
        Retrieve a particular feed identified by its id, identical to the source
21
        :return: JSON representation of the feed
22
        """
23
24
    @abc.abstractmethod
25
    def get_atom_feed_entry(self):
26
        """
27
        Retrieve a particular feed entry identified by its id, identical to the source
28
        :return: JSON representation of the feed entry
29
        """
30
31
    @abc.abstractmethod
32
    def archive_current_feed(self):
33
        """
34
        archive the current feed
35
        """
36
37
38
class AtomFeedView(AbstractAtomFeedView):
39
    """
40
    This is an implementation of the :class:`AbstractAtomFeedView` that adds a
41
    generic methods in a pylons pyramid application
42
    """
43
    __metaclass__ = abc.ABCMeta
44
45
    def __init__(self, request, atom_feed_manager, get_atom_feed_url, generate_atom_feed=None):
46
        """
47
        Functions that can be used in a pylons pyramid application
48
49
        :param request:
50
        :param atom_feed_manager: object of class oe_daemonutils.data.data_manager.AtomFeedManager
51
        :param get_atom_feed_url: the route name of the pyramid application to retrieve the atom feed
52
        """
53
        self.request = request
54
        self.atom_feed_manager = atom_feed_manager
55
        self.get_atom_feed_url = get_atom_feed_url
56
        self.generate_atom_feed = generate_atom_feed if generate_atom_feed else self._generate_atom_feed
57
58
    def get_current_atom_feed(self):
59
        return self.request.route_url(self.get_atom_feed_url, id=self.atom_feed_manager.current_feed.id)
60
61
    def get_atom_feed(self):
62
        feed_id = int(self.request.matchdict['id'])
63
        current_feed = self.atom_feed_manager.current_feed
64
        if feed_id == current_feed.id:
65
            feed = current_feed
66
            atom_feed = self.generate_atom_feed(feed).atom_str(pretty=True)
67
        else:
68
            atom_feed = self.atom_feed_manager.get_from_archive(feed_id)
69
        return atom_feed
70
71
    def get_atom_feed_entry(self):
72
        return self.atom_feed_manager.get_atom_feed_entry(int(self.request.matchdict['id']))
73
74
    def _archive_entire_feed(self, feed):
75
        current_atom_feed = self.generate_atom_feed(feed)
76
        new_feed = self.atom_feed_manager.save_new_feed()
77
        new_feed.uri = self.request.route_url(self.get_atom_feed_url, id=new_feed.id)
78
        current_atom_feed.link(
79
            href=new_feed.uri,
80
            rel='next-archive'
81
        )
82
        self.atom_feed_manager.archive_feed(feed.id, current_atom_feed)
83
84
    def _split_feed_and_archive(self, feed, split):
85
        leftover = feed.entries[split:]
86
        feed.entries = feed.entries[:split]
87
        current_atom_feed = self.generate_atom_feed(feed)
88
        new_feed = self.atom_feed_manager.feed_model()
89
        for leftover_entry in leftover:
90
            from sqlalchemy.orm import make_transient
91
            make_transient(leftover_entry)
92
            leftover_entry.feed = None
93
            leftover_entry.feed_id = None
94
        new_feed.entries = leftover
95
        new_feed = self.atom_feed_manager.save_object(new_feed)
96
        new_feed.uri = self.request.route_url(self.get_atom_feed_url, id=new_feed.id)
97
        current_atom_feed.link(
98
            href=new_feed.uri,
99
            rel='next-archive'
100
        )
101
        self.atom_feed_manager.archive_feed(feed.id, current_atom_feed)
102
        if len(new_feed.entries) > split:
103
            self._split_feed_and_archive(new_feed, split)
104
105
    def archive_current_feed(self):
106
        settings = self.request.registry.settings
107
        split = settings.get('feed.archive.split', None)
108
        current_feed = self.atom_feed_manager.current_feed
109
        if len(current_feed.entries) > 0:
110
            if split:
111
                split = int(split)
112
                self._split_feed_and_archive(current_feed, split)
113
            else:
114
                self._archive_entire_feed(current_feed)
115
116
    def link_to_sibling(self, feed, sibling_type, atom_feed):
117
        """
118
        Adding previous or next links to the given feed
119
        self._link_to_sibling(feed, 'previous', atom_feed)
120
        self._link_to_sibling(feed, 'next', atom_feed)
121
122
        :param feed: a feed object
123
        :param sibling_type: 'previous' or 'next'
124
        :param atom_feed: an atom feed like `feedgen.feed.FeedGenerator`
125
        """
126
        sibling = self.atom_feed_manager.get_sibling(feed.id, sibling_type)
127
        if sibling:
128
            rel = "prev-archive" if sibling_type == "previous" else "next-archive"
129
            atom_feed.link(href=self.request.route_url(self.get_atom_feed_url, id=sibling.id),
130
                           rel=rel)
131
132
    def init_atom_feed(self, feed):
133
        """
134
        Initializing an atom feed `feedgen.feed.FeedGenerator` given a feed object
135
136
        :param feed: a feed object
137
        :return: an atom feed `feedgen.feed.FeedGenerator`
138
        """
139
        atom_feed = FeedGenerator()
140
        atom_feed.id(id=self.request.route_url(self.get_atom_feed_url, id=feed.id))
141
        atom_feed.link(href=self.request.route_url(self.get_atom_feed_url, id=feed.id), rel='self')
142
        atom_feed.language('nl-BE')
143
        self.link_to_sibling(feed, 'previous', atom_feed)
144
        self.link_to_sibling(feed, 'next', atom_feed)
145
        return atom_feed
146
147
    def _generate_atom_feed(self, feed):
148
        """
149
        A function returning a feed like `feedgen.feed.FeedGenerator`.
150
        The function can be overwritten when used in other applications.
151
152
        :param feed: a feed object
153
        :return: an atom feed `feedgen.feed.FeedGenerator`
154
        """
155
        atom_feed = self.init_atom_feed(feed)
156
        atom_feed.title("Feed")
157
        return atom_feed
158