AtomFeedManager.get_atom_feed_entry() - Code Metrics - OnroerendErfgoed/oe_utils - Measure and Improve Code Quality continuously with Scrutinizer

AtomFeedManager.get_atom_feed_entry() A
last analyzed 2018-07-04 09:35 UTC

↳ Parent: AtomFeedManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	8
rs	10

# -*- coding: utf-8 -*-
from oe_utils.data.data_transfer_objects import ResultDTO
import os
from sqlalchemy.sql import desc
from oe_utils.data.models import Wijziging


class DataManager(object):
    """
    A datamanager base class.
    """

    @staticmethod
    def process_ranged_query(query, result_range):
        """

        :param query: the query to be processed
        :param result_range: :class: 'oe_utils.range_parser.Range'
        :return: :class:`oe_utils.data.data_transfer_objects.ResultDTO`
        """
        total = query.count()
        if result_range is not None:
            data = query \
                .offset(result_range.start) \
                .limit(result_range.get_page_size()) \
                .all()
        else:
            data = query.all()
        return ResultDTO(data, total)

    @staticmethod
    def ordered_result_list(query, model_type, sortlist):
        for sorttuple in sortlist:
            if hasattr(model_type, sorttuple[0]):
                attr = getattr(model_type, sorttuple[0])
                query = query.order_by(attr.asc()) if sorttuple[1] == 'asc' else query.order_by(attr.desc())
        return query

    def __init__(self, session, cls):
        """

        :param session: a db session
        :param cls: the class of the objects to manage
        :return:
        """
        self.session = session
        self.cls = cls

    def get_one(self, object_id):
        """
        Retrieve an object by its object_id

        :param object_id: the objects id.
        :return: the requested object
        :raises: :class: NoResultFound when the object could not be found
        """
        return self.session.query(self.cls).filter_by(id=object_id).one()

    def get_one_for_update(self, object_id):
        """
        Retrieve an object by its object_id
        Does a select for update on the row, results in row level lock for the duration of the transaction

        :param object_id: the objects id.
        :return: the requested object
        :raises: :class: NoResultFound when the object could not be found
        """
        return self.session.query(self.cls).with_for_update().filter_by(id=object_id).one()

    def get(self, object_id, cls=None):
        """
        Retrieve an object by its object_id

        :param: object_id: the objects id.
        :param: cls: the objects class, if None use the default class from the datamanager
        :return: the requested object or None if not found
        """
        cls = self.cls if cls is None else cls
        return self.session.query(cls).get(object_id)

    def get_for_update(self, object_id, cls=None):
        """
        Retrieve an object by its object_id
        Does a select for update on the row, results in row level lock for the duration of the transaction

        :param: object_id: the objects id.
        :param: cls: the objects class, if None use the default class from the datamanager
        :return: the requested object or None if not found
        """
        cls = self.cls if cls is None else cls
        return self.session.query(cls).with_for_update().get(object_id)

    def delete(self, object_id):
        """
        Delete an object by its id

        :param object_id: the objects id.
        :return: the deleted object
        :raises: :class: NoResultFound when the object could not be found
        """
        obj = self.session.query(self.cls).filter_by(id=object_id).one()
        self.session.delete(obj)
        return obj

    def save(self, obj):
        """
        save an object

        :param obj: the object
        :return: the saved object
        """
        if obj not in self.session:
            self.session.add(obj)
        else:
            obj = self.session.merge(obj)
        self.session.flush()
        self.session.refresh(obj)
        return obj


class FeedArchiveNotFound(Exception):
    pass


class AtomFeedManager(object):
    """
    A data manager for feeds.
    """
    def __init__(self, session, feed_repository, feed_model, feedentry_model):
        """
        :param session: a db session
        :param feed_repository: a repository to store the archived feeds
        """
        self.session = session
        self.feed_repository = feed_repository
        self.feed_model = feed_model
        self.feedentry_model = feedentry_model
        self._current_feed = None

    @property
    def current_feed(self):
        if self._current_feed is None:
            self._current_feed = self.session.query(self.feed_model).order_by(desc(self.feed_model.id)).first()
        return self._current_feed

    def get_sibling(self, feed_id, sibling_type):
        """
        get a previous/next sibling from a feed
        :param feed_id: id of the feed
        :param sibling_type: sibling type ('previous', 'next')
        :return: the sibling
        """
        if sibling_type == 'previous':
            query = self.session.query(self.feed_model).filter(self.feed_model.id < feed_id)
            order_clause = desc(self.feed_model.id)
        elif sibling_type == 'next':
            query = self.session.query(self.feed_model).filter(self.feed_model.id > feed_id)
            order_clause = self.feed_model.id
        else:
            raise Exception('Unhandled sibling relation type')  # pragma no cover
        if query.count() > 0:
            return query.order_by(order_clause).first()
        else:
            return None

    def save_new_feed(self):
        """
        Save a new feed object to the db
        :return: the saved object
        """
        obj = self.feed_model()
        return self.save_object(obj)

    def save_object(self, obj):
        """
        Save an object to the db
        :param obj: the object to save
        :return: the saved object
        """
        self.session.add(obj)
        self.session.flush()
        return obj

    def get_from_archive(self, feed_id):
        """
        Retrieves feed that was persisted as .xml file by its id (= filename)
        Note: No check on feed validity. file content is assumed correct
        :param feed_id:
        :return: the atom feed as string
        """
        file_path = self.feed_repository + '/' + str(feed_id) + '.xml'
        if not os.path.isfile(file_path):
            raise FeedArchiveNotFound()
        with open(file_path, 'r') as rec_file:
            return rec_file.read()

    def get_atom_feed_entry(self, feedentry_id):
        """
        Get a specific feed entry
        :param id: id of the feed entry to retrieve
        :return: the feed entry
        """
        return self.session.query(self.feedentry_model).filter(
            self.feedentry_model.id == feedentry_id
        ).one()

    def archive_feed(self, feed_id, feed):
        """
        Archive a feed
        :param feed_id: the feed id of the feed to archive
        :param feed: the feed to archive
        """
        with open(self.feed_repository + '/' + str(feed_id) + '.xml', 'w') as rec_file:
            rec_file.write(feed.atom_str(pretty=True))
        if feed_id == self.current_feed.id:
            self._current_feed = None


class AuditManager(DataManager):
    """
    A data manager for audit revisions.
    """

    def __init__(self, session):
        """
        :param session: a db session
        """
        super(AuditManager, self).__init__(session, Wijziging)
        self.session = session

    @staticmethod
    def create_revision():
        """
        Create a new revision object

        :return: new revision object
        """
        return Wijziging()


1			# -- coding: utf-8 --
2			from oe_utils.data.data_transfer_objects import ResultDTO
3			import os
4			from sqlalchemy.sql import desc
5			from oe_utils.data.models import Wijziging
6
7
8			class DataManager(object):
9			"""
10			A datamanager base class.
11			"""
12
13			@staticmethod
14			def process_ranged_query(query, result_range):
15			"""
16
17			:param query: the query to be processed
18			:param result_range: :class: 'oe_utils.range_parser.Range'
19			:return: :class:`oe_utils.data.data_transfer_objects.ResultDTO`
20			"""
21			total = query.count()
22			if result_range is not None:
23			data = query \
24			.offset(result_range.start) \
25			.limit(result_range.get_page_size()) \
26			.all()
27			else:
28			data = query.all()
29			return ResultDTO(data, total)
30
31			@staticmethod
32			def ordered_result_list(query, model_type, sortlist):
33			for sorttuple in sortlist:
34			if hasattr(model_type, sorttuple[0]):
35			attr = getattr(model_type, sorttuple[0])
36			query = query.order_by(attr.asc()) if sorttuple[1] == 'asc' else query.order_by(attr.desc())
37			return query
38
39			def __init__(self, session, cls):
40			"""
41
42			:param session: a db session
43			:param cls: the class of the objects to manage
44			:return:
45			"""
46			self.session = session
47			self.cls = cls
48
49			def get_one(self, object_id):
50			"""
51			Retrieve an object by its object_id
52
53			:param object_id: the objects id.
54			:return: the requested object
55			:raises: :class: NoResultFound when the object could not be found
56			"""
57			return self.session.query(self.cls).filter_by(id=object_id).one()
58
59			def get_one_for_update(self, object_id):
60			"""
61			Retrieve an object by its object_id
62			Does a select for update on the row, results in row level lock for the duration of the transaction
63
64			:param object_id: the objects id.
65			:return: the requested object
66			:raises: :class: NoResultFound when the object could not be found
67			"""
68			return self.session.query(self.cls).with_for_update().filter_by(id=object_id).one()
69
70			def get(self, object_id, cls=None):
71			"""
72			Retrieve an object by its object_id
73
74			:param: object_id: the objects id.
75			:param: cls: the objects class, if None use the default class from the datamanager
76			:return: the requested object or None if not found
77			"""
78			cls = self.cls if cls is None else cls
79			return self.session.query(cls).get(object_id)
80
81			def get_for_update(self, object_id, cls=None):
82			"""
83			Retrieve an object by its object_id
84			Does a select for update on the row, results in row level lock for the duration of the transaction
85
86			:param: object_id: the objects id.
87			:param: cls: the objects class, if None use the default class from the datamanager
88			:return: the requested object or None if not found
89			"""
90			cls = self.cls if cls is None else cls
91			return self.session.query(cls).with_for_update().get(object_id)
92
93			def delete(self, object_id):
94			"""
95			Delete an object by its id
96
97			:param object_id: the objects id.
98			:return: the deleted object
99			:raises: :class: NoResultFound when the object could not be found
100			"""
101			obj = self.session.query(self.cls).filter_by(id=object_id).one()
102			self.session.delete(obj)
103			return obj
104
105			def save(self, obj):
106			"""
107			save an object
108
109			:param obj: the object
110			:return: the saved object
111			"""
112			if obj not in self.session:
113			self.session.add(obj)
114			else:
115			obj = self.session.merge(obj)
116			self.session.flush()
117			self.session.refresh(obj)
118			return obj
119
120
121			class FeedArchiveNotFound(Exception):
122			pass
123
124
125			class AtomFeedManager(object):
126			"""
127			A data manager for feeds.
128			"""
129			def __init__(self, session, feed_repository, feed_model, feedentry_model):
130			"""
131			:param session: a db session
132			:param feed_repository: a repository to store the archived feeds
133			"""
134			self.session = session
135			self.feed_repository = feed_repository
136			self.feed_model = feed_model
137			self.feedentry_model = feedentry_model
138			self._current_feed = None
139
140			@property
141			def current_feed(self):
142			if self._current_feed is None:
143			self._current_feed = self.session.query(self.feed_model).order_by(desc(self.feed_model.id)).first()
144			return self._current_feed
145
146			def get_sibling(self, feed_id, sibling_type):
147			"""
148			get a previous/next sibling from a feed
149			:param feed_id: id of the feed
150			:param sibling_type: sibling type ('previous', 'next')
151			:return: the sibling
152			"""
153			if sibling_type == 'previous':
154			query = self.session.query(self.feed_model).filter(self.feed_model.id < feed_id)
155			order_clause = desc(self.feed_model.id)
156			elif sibling_type == 'next':
157			query = self.session.query(self.feed_model).filter(self.feed_model.id > feed_id)
158			order_clause = self.feed_model.id
159			else:
160			raise Exception('Unhandled sibling relation type') # pragma no cover
161			if query.count() > 0:
162			return query.order_by(order_clause).first()
163			else:
164			return None
165
166			def save_new_feed(self):
167			"""
168			Save a new feed object to the db
169			:return: the saved object
170			"""
171			obj = self.feed_model()
172			return self.save_object(obj)
173
174			def save_object(self, obj):
175			"""
176			Save an object to the db
177			:param obj: the object to save
178			:return: the saved object
179			"""
180			self.session.add(obj)
181			self.session.flush()
182			return obj
183
184			def get_from_archive(self, feed_id):
185			"""
186			Retrieves feed that was persisted as .xml file by its id (= filename)
187			Note: No check on feed validity. file content is assumed correct
188			:param feed_id:
189			:return: the atom feed as string
190			"""
191			file_path = self.feed_repository + '/' + str(feed_id) + '.xml'
192			if not os.path.isfile(file_path):
193			raise FeedArchiveNotFound()
194			with open(file_path, 'r') as rec_file:
195			return rec_file.read()
196
197			def get_atom_feed_entry(self, feedentry_id):
198			"""
199			Get a specific feed entry
200			:param id: id of the feed entry to retrieve
201			:return: the feed entry
202			"""
203			return self.session.query(self.feedentry_model).filter(
204			self.feedentry_model.id == feedentry_id
205			).one()
206
207			def archive_feed(self, feed_id, feed):
208			"""
209			Archive a feed
210			:param feed_id: the feed id of the feed to archive
211			:param feed: the feed to archive
212			"""
213			with open(self.feed_repository + '/' + str(feed_id) + '.xml', 'w') as rec_file:
214			rec_file.write(feed.atom_str(pretty=True))
215			if feed_id == self.current_feed.id:
216			self._current_feed = None
217
218
219			class AuditManager(DataManager):
220			"""
221			A data manager for audit revisions.
222			"""
223
224			def __init__(self, session):
225			"""
226			:param session: a db session
227			"""
228			super(AuditManager, self).__init__(session, Wijziging)
229			self.session = session
230
231			@staticmethod
232			def create_revision():
233			"""
234			Create a new revision object
235
236			:return: new revision object
237			"""
238			return Wijziging()
239

OnroerendErfgoed / oe_utils

AtomFeedManager.get_atom_feed_entry() A last analyzed 2018-07-04 09:35 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

AtomFeedManager.get_atom_feed_entry() A
last analyzed 2018-07-04 09:35 UTC