AuditManager.__init__() - Code Metrics - Inspection of "Merge pull request #21 from OnroerendErfgoed/audit" - OnroerendErfgoed/oe_utils - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 78de89...99647d )

unknown

created 2016-08-03 07:09 UTC

AuditManager.init() A

↳ Parent: AuditManager

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	6
rs	9.4285

# -*- 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)

    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(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 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			def __init__(self, session, cls):
32			"""
33
34			:param session: a db session
35			:param cls: the class of the objects to manage
36			:return:
37			"""
38			self.session = session
39			self.cls = cls
40
41			def get_one(self, object_id):
42			"""
43			Retrieve an object by its object_id
44
45			:param object_id: the objects id.
46			:return: the requested object
47			:raises: :class: NoResultFound when the object could not be found
48			"""
49			return self.session.query(self.cls).filter_by(id=object_id).one()
50
51			def get(self, object_id, cls=None):
52			"""
53			Retrieve an object by its object_id
54
55			:param: object_id: the objects id.
56			:param: cls: the objects class, if None use the default class from the datamanager
57			:return: the requested object or None if not found
58			"""
59			cls = self.cls if cls is None else cls
60			return self.session.query(cls).get(object_id)
61
62			def delete(self, object_id):
63			"""
64			Delete an object by its id
65
66			:param object_id: the objects id.
67			:return: the deleted object
68			:raises: :class: NoResultFound when the object could not be found
69			"""
70			obj = self.session.query(self.cls).filter_by(id=object_id).one()
71			self.session.delete(obj)
72			return obj
73
74			def save(self, obj):
75			"""
76			save an object
77
78			:param obj: the object
79			:return: the saved object
80			"""
81			if obj not in self.session:
82			self.session.add(obj)
83			else:
84			obj = self.session.merge(obj)
85			self.session.flush()
86			self.session.refresh(obj)
87			return obj
88
89
90			class FeedArchiveNotFound(Exception):
91			pass
92
93
94			class AtomFeedManager(object):
95			"""
96			A data manager for feeds.
97			"""
98			def __init__(self, session, feed_repository, feed_model, feedentry_model):
99			"""
100			:param session: a db session
101			:param feed_repository: a repository to store the archived feeds
102			"""
103			self.session = session
104			self.feed_repository = feed_repository
105			self.feed_model = feed_model
106			self.feedentry_model = feedentry_model
107			self._current_feed = None
108
109			@property
110			def current_feed(self):
111			if self._current_feed is None:
112			self._current_feed = self.session.query(self.feed_model).order_by(desc(self.feed_model.id)).first()
113			return self._current_feed
114
115			def get_sibling(self, feed_id, sibling_type):
116			"""
117			get a previous/next sibling from a feed
118			:param feed_id: id of the feed
119			:param sibling_type: sibling type ('previous', 'next')
120			:return: the sibling
121			"""
122			if sibling_type == 'previous':
123			query = self.session.query(self.feed_model).filter(self.feed_model.id < feed_id)
124			order_clause = desc(self.feed_model.id)
125			elif sibling_type == 'next':
126			query = self.session.query(self.feed_model).filter(self.feed_model.id > feed_id)
127			order_clause = self.feed_model.id
128			else:
129			raise Exception('Unhandled sibling relation type') # pragma no cover
130			if query.count() > 0:
131			return query.order_by(order_clause).first()
132			else:
133			return None
134
135			def save_new_feed(self):
136			"""
137			Save a new feed object to the db
138			:return: the saved object
139			"""
140			obj = self.feed_model()
141			return self.save_object(obj)
142
143			def save_object(self, obj):
144			"""
145			Save an object to the db
146			:param obj: the object to save
147			:return: the saved object
148			"""
149			self.session.add(obj)
150			self.session.flush()
151			return obj
152
153			def get_from_archive(self, feed_id):
154			"""
155			Retrieves feed that was persisted as .xml file by its id (= filename)
156			Note: No check on feed validity. file content is assumed correct
157			:param feed_id:
158			:return: the atom feed as string
159			"""
160			file_path = self.feed_repository + '/' + str(feed_id) + '.xml'
161			if not os.path.isfile(file_path):
162			raise FeedArchiveNotFound()
163			with open(file_path, 'r') as rec_file:
164			return rec_file.read()
165
166			def get_atom_feed_entry(self, feedentry_id):
167			"""
168			Get a specific feed entry
169			:param id: id of the feed entry to retrieve
170			:return: the feed entry
171			"""
172			return self.session.query(self.feedentry_model).filter(
173			self.feedentry_model.id == feedentry_id
174			).one()
175
176			def archive_feed(self, feed_id, feed):
177			"""
178			Archive a feed
179			:param feed_id: the feed id of the feed to archive
180			:param feed: the feed to archive
181			"""
182			with open(self.feed_repository + '/' + str(feed_id) + '.xml', 'w') as rec_file:
183			rec_file.write(feed.atom_str(pretty=True))
184			if feed_id == self.current_feed.id:
185			self._current_feed = None
186
187
188			class AuditManager(DataManager):
189			"""
190			A data manager for audit revisions.
191			"""
192
193			def __init__(self, session):
194			"""
195			:param session: a db session
196			"""
197			super(AuditManager, self).__init__(session, Wijziging)
198			self.session = session
199
200			@staticmethod
201			def create_revision():
202			"""
203			Create a new revision object
204
205			:return: new revision object
206			"""
207			return Wijziging()
208

OnroerendErfgoed / oe_utils

Push — master ( 78de89...99647d )

AuditManager.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

AuditManager.init() A