atramhasis.data.datamanagers.SkosManager.get_next_cid() - Code Metrics - OnroerendErfgoed/atramhasis - Measure and Improve Code Quality continuously with Scrutinizer

SkosManager.get_next_cid() A
last analyzed 2024-08-12 13:03 UTC

↳ Parent: atramhasis.data.datamanagers

Complexity

Conditions

Size

Total Lines	11
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

OnroerendErfgoed / atramhasis

SkosManager.get_next_cid() A last analyzed 2024-08-12 13:03 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

SkosManager.get_next_cid() A
last analyzed 2024-08-12 13:03 UTC