atramhasis.data.datamanagers.ConceptSchemeManager.get() - Code Metrics - Inspection of "Merge branch 'develop'" - OnroerendErfgoed/atramhasis - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 9ea4cd...0ffab9 )

by Koen

created 2023-01-04 14:55 UTC

ConceptSchemeManager.get() A

↳ Parent: atramhasis.data.datamanagers

Complexity

Conditions

Size

Total Lines	7
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	7
rs	10
c	0
b	0
f	0
cc	1
nop	2

"""
This module adds DataManagers for Atramhasis. These are service layer objects
that abstract all interactions with the database away from the views.

:versionadded: 0.4.1
"""
from datetime import date
from datetime import datetime

import dateutil.relativedelta
from skosprovider_sqlalchemy.models import Collection
from skosprovider_sqlalchemy.models import Concept
from skosprovider_sqlalchemy.models import ConceptScheme
from skosprovider_sqlalchemy.models import Label
from skosprovider_sqlalchemy.models import LabelType
from skosprovider_sqlalchemy.models import Language
from skosprovider_sqlalchemy.models import Match
from skosprovider_sqlalchemy.models import MatchType
from skosprovider_sqlalchemy.models import Thing
from sqlalchemy import and_
from sqlalchemy import desc
from sqlalchemy import func
from sqlalchemy.orm import joinedload

from atramhasis.data import popular_concepts
from atramhasis.data.models import ConceptVisitLog
from atramhasis.data.models import ConceptschemeCounts


class DataManager:
    """
    A DataManager abstracts all interactions with the database for a certain model.
    """

    def __init__(self, session):
        self.session = session


class ConceptSchemeManager(DataManager):
    """
    A :class:`DataManager` for
    :class:`ConceptSchemes <skosprovider_sqlalchemy.models.ConceptScheme>.`
    """

    def __init__(self, session):
        super().__init__(session)

    def get(self, conceptscheme_id):
        """

        :param conceptscheme_id: a concepscheme id
        :return: the concepscheme for the given id
        """
        return self.session.query(ConceptScheme).filter_by(id=conceptscheme_id).one()

    def find(self, conceptscheme_id, query):
        """
        Find concepts and collections in this concept scheme.

        :param conceptscheme_id: a concepscheme id
        :param query: A python dictionary containing query parameters.
        :returns: A :class:`list` of
            :class:`skosprovider_sqlalchemy.models.Thing` instances.
        """
        q = self.session \
            .query(Thing) \
            .options(joinedload('labels')) \
            .filter(Thing.conceptscheme_id == conceptscheme_id)
        if 'type' in query and query['type'] in ['concept', 'collection']:
            q = q.filter(Thing.type == query['type'])
        if 'label' in query:
            q = q.filter(
                Thing.labels.any(
                    Label.label.ilike('%' + query['label'].lower() + '%')
                )
            )
        return q.all()

    def get_concepts_for_scheme_tree(self, conceptscheme_id):
        """

        :param conceptscheme_id:  a concepscheme id
        :return: all concepts for the scheme_tree
        """
        return self.session \
            .query(Concept) \
            .filter(Concept.conceptscheme_id == conceptscheme_id,
                    ~Concept.broader_concepts.any(),
                    ~Collection.member_of.any()
                    ).all()

    def get_collections_for_scheme_tree(self, conceptscheme_id):
        """

        :param conceptscheme_id: a concepscheme id
        :return: all collections for the scheme_tree
        """
        return self.session \
            .query(Collection) \
            .filter(Collection.conceptscheme_id == conceptscheme_id,
                    ~Collection.broader_concepts.any(),
                    ~Collection.member_of.any()
                    ).all()

    def get_all(self, conceptscheme_id):
        """
        Get all concepts and collections in this concept scheme.

        :param conceptscheme_id: a concepscheme id
        :returns: A :class:`list` of
            :class:`skosprovider_sqlalchemy.models.Thing` instances.
        """
        all_results = self.session \
            .query(Thing) \
            .options(joinedload('labels')) \
            .filter(Thing.conceptscheme_id == conceptscheme_id) \
            .all()
        return all_results

    def save(self, conceptscheme):
        """

        :param conceptscheme: conceptscheme to save
        :return: saved conceptscheme
        """
        self.session.merge(conceptscheme)
        self.session.flush()
        return conceptscheme


class SkosManager(DataManager):
    """
    A :class:`DataManager` for
    :class:`Concepts and Collections <skosprovider_sqlalchemy.models.Thing>.`
    """

    def __init__(self, session):
        super().__init__(session)

    def get_thing(self, concept_id, conceptscheme_id):
        """

        :param concept_id: a concept id
        :param conceptscheme_id: a conceptscheme id
        :return: the selected thing (Concept or Collection)
        """
        return self.session.query(Thing) \
            .filter_by(concept_id=concept_id, conceptscheme_id=conceptscheme_id) \
            .one()

    def save(self, thing):
        """

        :param thing: thing to save
        :return: saved thing
        """
        self.session.add(thing)
        self.session.flush()
        return thing

    def change_type(self, thing, concept_id, conceptscheme_id, new_type, uri):
        self.delete_thing(thing)
        self.session.flush()
        thing = Concept() if new_type == 'concept' else Collection()
        thing.type = new_type
        thing.concept_id = concept_id
        thing.conceptscheme_id = conceptscheme_id
        thing.uri = uri
        self.save(thing)
        return thing

    def delete_thing(self, thing):
        """

        :param thing: the thing to delete
        """
        self.session.delete(thing)

    def get_by_list_type(self, list_type):
        """

        :param list_type: a specific list type
        :return: all results for the specific list type
        """
        return self.session.query(list_type).all()

    def get_match_type(self, match_type):
        return self.session.query(MatchType).filter_by(name=match_type).one()

    def get_match(self, uri, matchtype_id, concept_id):
        return self.session.query(Match).filter_by(uri=uri, matchtype_id=matchtype_id,
                                                   concept_id=concept_id).one()

    def get_all_label_types(self):
        return self.session.query(LabelType).all()

    def get_next_cid(self, conceptscheme_id):
        max_id = self.session.query(
            func.max(Thing.concept_id)
        ).filter_by(conceptscheme_id=conceptscheme_id).first()[0]
        return max_id + 1 if max_id else 1


class LanguagesManager(DataManager):
    """
    A :class:`DataManager` for
    :class:`Languages <skosprovider_sqlalchemy.models.Language>.`
    """

    def __init__(self, session):
        super().__init__(session)

    def get(self, language_id):
        return self.session.query(Language).filter_by(id=language_id).one()

    def save(self, language):
        """

        :param language: language to save
        :return: saved language
        """
        self.session.add(language)
        self.session.flush()
        return language

    def delete(self, language):
        """

        :param language: the language to delete
        """
        self.session.delete(language)

    def get_all(self):
        """

        :return: list of all languages
        """
        return self.session.query(Language).all()

    def get_all_sorted(self, sort_coll, sort_desc):
        """

        :param sort_coll: sort on this column
        :param sort_desc: descending or not
        :return: sorted list of languages
        """
        if sort_desc:
            languages = self.session.query(Language).order_by(desc(sort_coll)).all()
        else:
            languages = self.session.query(Language).order_by(sort_coll).all()
        return languages

    def count_languages(self, language_tag):
        return self.session.query(Language).filter_by(id=language_tag).count()


class AuditManager(DataManager):
    """
    A data manager for logging the visit.
    """

    def save(self, visit_log):
        """
        save a certain visit
        :param visit_log: log of visit to save
        :return: The saved visit log
        """
        self.session.add(visit_log)
        self.session.flush()
        return visit_log

    @popular_concepts.cache_on_arguments(expiration_time=86400)
    def get_most_popular_concepts_for_conceptscheme(self, conceptscheme_id, max=5, period='last_month'):
        """
        get the most popular concepts for a conceptscheme
        :param conceptscheme_id: id of the conceptscheme
        :param max: maximum number of results, default 5
        :param period: 'last_day' or 'last_week' or 'last_month' or 'last_year', default 'last_month'
        :return: List of the most popular concepts of a conceptscheme over a certain period
        """

        start_date = self._get_first_day(period)
        popular_concepts = self.session.query(
            ConceptVisitLog.concept_id,
            func.count(ConceptVisitLog.concept_id).label('count')
        ).filter(
            and_(ConceptVisitLog.conceptscheme_id == str(conceptscheme_id),
                 ConceptVisitLog.visited_at >= start_date)
        ).group_by(
            ConceptVisitLog.concept_id
        ).order_by(
            desc('count')
        ).limit(
            max
        ).all()
        results = []
        for concept in popular_concepts:
            results.append({'concept_id': concept.concept_id, 'scheme_id': conceptscheme_id})
        return results

    @staticmethod
    def _get_first_day(period):
        """
        get the first day of a certain period until now
        :param period: 'last_day' or 'last_week' or 'last_month' or 'last_year'
        :return: (string) the first day of the period
        """
        d = date.today()
        datetime.combine(d, datetime.min.time())
        start_date = d - dateutil.relativedelta.relativedelta(
            days=1 if period == 'last_day' else 0,
            weeks=1 if period == 'last_week' else 0,
            months=1 if period == 'last_month' else 0,
            years=1 if period == 'last_year' else 0
        )
        return start_date.strftime("%Y-%m-%d")


class CountsManager(DataManager):
    """
    A data manager that deals with triple counts.
    """

    def save(self, counts):
        """
        Save a certain counts object

        :param atramhasis.data.models.ConceptschemeCounts counts: Counts object to save

        :return: The saved count
        """
        self.session.add(counts)
        self.session.flush()
        return counts

    def get_most_recent_count_for_scheme(self, conceptscheme_id):
        recent = self.session.query(
            ConceptschemeCounts
        ).filter_by(
            conceptscheme_id=conceptscheme_id
        ).order_by(
            desc('counted_at')
        ).one()
        return recent


1			"""
2			This module adds DataManagers for Atramhasis. These are service layer objects
3			that abstract all interactions with the database away from the views.
4
5			:versionadded: 0.4.1
6			"""
7			from datetime import date
8			from datetime import datetime
9
10			import dateutil.relativedelta
11			from skosprovider_sqlalchemy.models import Collection
12			from skosprovider_sqlalchemy.models import Concept
13			from skosprovider_sqlalchemy.models import ConceptScheme
14			from skosprovider_sqlalchemy.models import Label
15			from skosprovider_sqlalchemy.models import LabelType
16			from skosprovider_sqlalchemy.models import Language
17			from skosprovider_sqlalchemy.models import Match
18			from skosprovider_sqlalchemy.models import MatchType
19			from skosprovider_sqlalchemy.models import Thing
20			from sqlalchemy import and_
21			from sqlalchemy import desc
22			from sqlalchemy import func
23			from sqlalchemy.orm import joinedload
24
25			from atramhasis.data import popular_concepts
26			from atramhasis.data.models import ConceptVisitLog
27			from atramhasis.data.models import ConceptschemeCounts
28
29
30			class DataManager:
31			"""
32			A DataManager abstracts all interactions with the database for a certain model.
33			"""
34
35			def __init__(self, session):
36			self.session = session
37
38
39			class ConceptSchemeManager(DataManager):
40			"""
41			A :class:`DataManager` for
42			:class:`ConceptSchemes <skosprovider_sqlalchemy.models.ConceptScheme>.`
43			"""
44
45			def __init__(self, session):
46			super().__init__(session)
47
48			def get(self, conceptscheme_id):
49			"""
50
51			:param conceptscheme_id: a concepscheme id
52			:return: the concepscheme for the given id
53			"""
54			return self.session.query(ConceptScheme).filter_by(id=conceptscheme_id).one()
55
56			def find(self, conceptscheme_id, query):
57			"""
58			Find concepts and collections in this concept scheme.
59
60			:param conceptscheme_id: a concepscheme id
61			:param query: A python dictionary containing query parameters.
62			:returns: A :class:`list` of
63			:class:`skosprovider_sqlalchemy.models.Thing` instances.
64			"""
65			q = self.session \
66			.query(Thing) \
67			.options(joinedload('labels')) \
68			.filter(Thing.conceptscheme_id == conceptscheme_id)
69			if 'type' in query and query['type'] in ['concept', 'collection']:
70			q = q.filter(Thing.type == query['type'])
71			if 'label' in query:
72			q = q.filter(
73			Thing.labels.any(
74			Label.label.ilike('%' + query['label'].lower() + '%')
75			)
76			)
77			return q.all()
78
79			def get_concepts_for_scheme_tree(self, conceptscheme_id):
80			"""
81
82			:param conceptscheme_id: a concepscheme id
83			:return: all concepts for the scheme_tree
84			"""
85			return self.session \
86			.query(Concept) \
87			.filter(Concept.conceptscheme_id == conceptscheme_id,
88			~Concept.broader_concepts.any(),
89			~Collection.member_of.any()
90			).all()
91
92			def get_collections_for_scheme_tree(self, conceptscheme_id):
93			"""
94
95			:param conceptscheme_id: a concepscheme id
96			:return: all collections for the scheme_tree
97			"""
98			return self.session \
99			.query(Collection) \
100			.filter(Collection.conceptscheme_id == conceptscheme_id,
101			~Collection.broader_concepts.any(),
102			~Collection.member_of.any()
103			).all()
104
105			def get_all(self, conceptscheme_id):
106			"""
107			Get all concepts and collections in this concept scheme.
108
109			:param conceptscheme_id: a concepscheme id
110			:returns: A :class:`list` of
111			:class:`skosprovider_sqlalchemy.models.Thing` instances.
112			"""
113			all_results = self.session \
114			.query(Thing) \
115			.options(joinedload('labels')) \
116			.filter(Thing.conceptscheme_id == conceptscheme_id) \
117			.all()
118			return all_results
119
120			def save(self, conceptscheme):
121			"""
122
123			:param conceptscheme: conceptscheme to save
124			:return: saved conceptscheme
125			"""
126			self.session.merge(conceptscheme)
127			self.session.flush()
128			return conceptscheme
129
130
131			class SkosManager(DataManager):
132			"""
133			A :class:`DataManager` for
134			:class:`Concepts and Collections <skosprovider_sqlalchemy.models.Thing>.`
135			"""
136
137			def __init__(self, session):
138			super().__init__(session)
139
140			def get_thing(self, concept_id, conceptscheme_id):
141			"""
142
143			:param concept_id: a concept id
144			:param conceptscheme_id: a conceptscheme id
145			:return: the selected thing (Concept or Collection)
146			"""
147			return self.session.query(Thing) \
148			.filter_by(concept_id=concept_id, conceptscheme_id=conceptscheme_id) \
149			.one()
150
151			def save(self, thing):
152			"""
153
154			:param thing: thing to save
155			:return: saved thing
156			"""
157			self.session.add(thing)
158			self.session.flush()
159			return thing
160
161			def change_type(self, thing, concept_id, conceptscheme_id, new_type, uri):
162			self.delete_thing(thing)
163			self.session.flush()
164			thing = Concept() if new_type == 'concept' else Collection()
165			thing.type = new_type
166			thing.concept_id = concept_id
167			thing.conceptscheme_id = conceptscheme_id
168			thing.uri = uri
169			self.save(thing)
170			return thing
171
172			def delete_thing(self, thing):
173			"""
174
175			:param thing: the thing to delete
176			"""
177			self.session.delete(thing)
178
179			def get_by_list_type(self, list_type):
180			"""
181
182			:param list_type: a specific list type
183			:return: all results for the specific list type
184			"""
185			return self.session.query(list_type).all()
186
187			def get_match_type(self, match_type):
188			return self.session.query(MatchType).filter_by(name=match_type).one()
189
190			def get_match(self, uri, matchtype_id, concept_id):
191			return self.session.query(Match).filter_by(uri=uri, matchtype_id=matchtype_id,
192			concept_id=concept_id).one()
193
194			def get_all_label_types(self):
195			return self.session.query(LabelType).all()
196
197			def get_next_cid(self, conceptscheme_id):
198			max_id = self.session.query(
199			func.max(Thing.concept_id)
200			).filter_by(conceptscheme_id=conceptscheme_id).first()[0]
201			return max_id + 1 if max_id else 1
202
203
204			class LanguagesManager(DataManager):
205			"""
206			A :class:`DataManager` for
207			:class:`Languages <skosprovider_sqlalchemy.models.Language>.`
208			"""
209
210			def __init__(self, session):
211			super().__init__(session)
212
213			def get(self, language_id):
214			return self.session.query(Language).filter_by(id=language_id).one()
215
216			def save(self, language):
217			"""
218
219			:param language: language to save
220			:return: saved language
221			"""
222			self.session.add(language)
223			self.session.flush()
224			return language
225
226			def delete(self, language):
227			"""
228
229			:param language: the language to delete
230			"""
231			self.session.delete(language)
232
233			def get_all(self):
234			"""
235
236			:return: list of all languages
237			"""
238			return self.session.query(Language).all()
239
240			def get_all_sorted(self, sort_coll, sort_desc):
241			"""
242
243			:param sort_coll: sort on this column
244			:param sort_desc: descending or not
245			:return: sorted list of languages
246			"""
247			if sort_desc:
248			languages = self.session.query(Language).order_by(desc(sort_coll)).all()
249			else:
250			languages = self.session.query(Language).order_by(sort_coll).all()
251			return languages
252
253			def count_languages(self, language_tag):
254			return self.session.query(Language).filter_by(id=language_tag).count()
255
256
257			class AuditManager(DataManager):
258			"""
259			A data manager for logging the visit.
260			"""
261
262			def save(self, visit_log):
263			"""
264			save a certain visit
265			:param visit_log: log of visit to save
266			:return: The saved visit log
267			"""
268			self.session.add(visit_log)
269			self.session.flush()
270			return visit_log
271
272			@popular_concepts.cache_on_arguments(expiration_time=86400)
273			def get_most_popular_concepts_for_conceptscheme(self, conceptscheme_id, max=5, period='last_month'):
274			"""
275			get the most popular concepts for a conceptscheme
276			:param conceptscheme_id: id of the conceptscheme
277			:param max: maximum number of results, default 5
278			:param period: 'last_day' or 'last_week' or 'last_month' or 'last_year', default 'last_month'
279			:return: List of the most popular concepts of a conceptscheme over a certain period
280			"""
281
282			start_date = self._get_first_day(period)
283			popular_concepts = self.session.query(
284			ConceptVisitLog.concept_id,
285			func.count(ConceptVisitLog.concept_id).label('count')
286			).filter(
287			and_(ConceptVisitLog.conceptscheme_id == str(conceptscheme_id),
288			ConceptVisitLog.visited_at >= start_date)
289			).group_by(
290			ConceptVisitLog.concept_id
291			).order_by(
292			desc('count')
293			).limit(
294			max
295			).all()
296			results = []
297			for concept in popular_concepts:
298			results.append({'concept_id': concept.concept_id, 'scheme_id': conceptscheme_id})
299			return results
300
301			@staticmethod
302			def _get_first_day(period):
303			"""
304			get the first day of a certain period until now
305			:param period: 'last_day' or 'last_week' or 'last_month' or 'last_year'
306			:return: (string) the first day of the period
307			"""
308			d = date.today()
309			datetime.combine(d, datetime.min.time())
310			start_date = d - dateutil.relativedelta.relativedelta(
311			days=1 if period == 'last_day' else 0,
312			weeks=1 if period == 'last_week' else 0,
313			months=1 if period == 'last_month' else 0,
314			years=1 if period == 'last_year' else 0
315			)
316			return start_date.strftime("%Y-%m-%d")
317
318
319			class CountsManager(DataManager):
320			"""
321			A data manager that deals with triple counts.
322			"""
323
324			def save(self, counts):
325			"""
326			Save a certain counts object
327
328			:param atramhasis.data.models.ConceptschemeCounts counts: Counts object to save
329
330			:return: The saved count
331			"""
332			self.session.add(counts)
333			self.session.flush()
334			return counts
335
336			def get_most_recent_count_for_scheme(self, conceptscheme_id):
337			recent = self.session.query(
338			ConceptschemeCounts
339			).filter_by(
340			conceptscheme_id=conceptscheme_id
341			).order_by(
342			desc('counted_at')
343			).one()
344			return recent
345

OnroerendErfgoed / atramhasis

Push — master ( 9ea4cd...0ffab9 )

ConceptSchemeManager.get() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like