atramhasis.data.datamanagers.AuditManager._get_first_day() - Code Metrics - Inspection of "Merge pull request #765 from OnroerendErfgoed/epic..." - OnroerendErfgoed/atramhasis - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 75ee4b...de0baf )

by Jonas

created 2023-01-31 09:57 UTC

AuditManager._get_first_day() A

↳ Parent: atramhasis.data.datamanagers

Complexity

Conditions

Size

Total Lines	16
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	10
dl	0
loc	16
rs	9.3333
c	0
b	0
f	0
cc	5
nop	1

"""
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
"""
import uuid

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 desc
from sqlalchemy import func
from sqlalchemy import select
from sqlalchemy.orm import joinedload

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


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.execute(
            select(ConceptScheme)
            .filter(ConceptScheme.id == conceptscheme_id)
        ).scalar_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.
        """
        db_query = (
            select(Thing)
            .options(joinedload('labels'))
            .filter(Thing.conceptscheme_id == conceptscheme_id)
        )
        if 'type' in query and query['type'] in ['concept', 'collection']:
            db_query = db_query.filter(Thing.type == query['type'])
        if 'label' in query:
            db_query = db_query.filter(
                Thing.labels.any(
                    Label.label.ilike('%' + query['label'].lower() + '%')
                )
            )
        return self.session.execute(db_query).unique().scalars().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.execute(
            select(Concept)
            .filter(
                Concept.conceptscheme_id == conceptscheme_id,
                ~Concept.broader_concepts.any(),
                ~Collection.member_of.any()
            )
        ).scalars().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.execute(
            select(Collection)
            .filter(
                Collection.conceptscheme_id == conceptscheme_id,
                ~Collection.broader_concepts.any(),
                ~Collection.member_of.any(),
            )
        ).scalars().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.execute(
            select(Thing)
            .options(joinedload('labels'))
            .filter(Thing.conceptscheme_id == conceptscheme_id)
        ).unique().scalars().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.execute(
            select(Thing)
            .filter(
                Thing.concept_id == concept_id,
                Thing.conceptscheme_id == conceptscheme_id
            )
        ).scalar_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.execute(select(list_type)).scalars().all()

    def get_match_type(self, match_type):
        return self.session.execute(
            select(MatchType)
            .filter(MatchType.name == match_type)
        ).scalar_one()

    def get_match(self, uri, matchtype_id, concept_id):
        return self.session.execute(
            select(Match)
            .filter(
                Match.uri == uri,
                Match.matchtype_id == matchtype_id,
                Match.concept_id == concept_id
            )
        ).scalar_one()

    def get_all_label_types(self):
        return self.session.execute(select(LabelType)).scalars().all()

    def get_next_cid(self, conceptscheme_id, id_generation_strategy):
        if id_generation_strategy == IDGenerationStrategy.NUMERIC:
            max_id = self.session.execute(
               select(func.max(Thing.concept_id))
               .filter_by(conceptscheme_id=conceptscheme_id)
            ).scalar_one()
            return max_id + 1 if max_id else 1
        elif id_generation_strategy == IDGenerationStrategy.GUID:
            return str(uuid.uuid4())
        else:
            raise ValueError("unsupported id_generation_strategy")


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.execute(
            select(Language)
            .filter(Language.id == language_id)
        ).scalar_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.execute(select(Language)).scalars().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:
            return self.session.execute(
                select(Language)
                .order_by(desc(sort_coll))
            ).scalars().all()
        else:
            return self.session.execute(
                select(Language)
                .order_by(sort_coll)
            ).scalars().all()

    def count_languages(self, language_tag):
        return self.session.execute(
            select(func.count(Language.id))
            .filter(Language.id == language_tag)
        ).scalar_one()


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_results=5, period='last_month'
    ):
        """
        get the most popular concepts for a conceptscheme
        :param conceptscheme_id: id of the conceptscheme
        :param max_results: maximum number of results, default 5
        :param period: 'last_day' or 'last_week' or 'last_month' or 'last_year', default 'last_mont h'
        :return: List of the most popular concepts of a conceptscheme over a certain period
        """

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

OnroerendErfgoed / atramhasis

Push — develop ( 75ee4b...de0baf )

AuditManager._get_first_day() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like