atramhasis.data.datamanagers.SkosManager.get_thing() - Code Metrics - Inspection of "Add Create, Update, Delete provider." - OnroerendErfgoed/atramhasis - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — develop (#812)

unknown

created 2023-04-13 06:26 UTC

SkosManager.get_thing() A

↳ Parent: atramhasis.data.datamanagers

Complexity

Conditions

Size

Total Lines	14
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

"""
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
from typing import List

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 Session
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.data.models import IDGenerationStrategy
from atramhasis.data.models import Provider
from atramhasis.scripts import delete_scheme


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

    def __init__(self, session: Session) -> None:
        self.session: 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


class ProviderDataManager(DataManager):
    """A data manager for managing Providers."""

    def get_provider_by_id(self, provider_id) -> Provider:
        return self.session.execute(
            select(Provider)
            .filter(Provider.id == provider_id)
        ).scalar_one()

    def get_all_providers(self) -> List[Provider]:
        """
        Retrieve all providers from the database.

        :return: All providers
        """
        return self.session.execute(select(Provider)).scalars().all()


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

OnroerendErfgoed / atramhasis

Pull Request — develop (#812)

SkosManager.get_thing() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like