atramhasis.utils.from_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-03-22 15:31 UTC

atramhasis.utils.from_thing() B

↳ Parent: atramhasis.utils

Complexity

Conditions

Size

Total Lines	61
Code Lines	44

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	44
dl	0
loc	61
rs	7.8906
c	0
b	0
f	0
cc	6
nop	1

How to fix Long Method

"""
Module containing utility functions used by Atramhasis.
"""
import copy
from collections import deque

from pyramid.httpexceptions import HTTPMethodNotAllowed
from skosprovider.skos import Collection
from skosprovider.skos import Concept
from skosprovider.skos import ConceptScheme
from skosprovider.skos import Label
from skosprovider.skos import Note
from skosprovider.skos import Source
from skosprovider.uri import UriPatternGenerator
from skosprovider_sqlalchemy.providers import SQLAlchemyProvider
from sqlalchemy import orm
from sqlalchemy.orm import Session

from atramhasis.data.models import IDGenerationStrategy
from atramhasis.data.models import Provider


def from_thing(thing):
    """
        Map a :class:`skosprovider_sqlalchemy.models.Thing` to a
        :class:`skosprovider.skos.Concept` or
        a :class:`skosprovider.skos.Collection`, depending on the type.

        :param skosprovider_sqlalchemy.models.Thing thing: Thing to map.
        :rtype: :class:`~skosprovider.skos.Concept` or
            :class:`~skosprovider.skos.Collection`.
        """
    if thing.type and thing.type == 'collection':
        return Collection(
            id=thing.concept_id,
            uri=thing.uri,
            concept_scheme=ConceptScheme(thing.conceptscheme.uri),
            labels=[
                Label(label.label, label.labeltype_id, label.language_id)
                for label in thing.labels
            ],
            notes=[
                Note(n.note, n.notetype_id, n.language_id)
                for n in thing.notes
            ],
            sources=[
                Source(s.citation)
                for s in thing.sources
            ],
            members=[member.concept_id for member in thing.members] if hasattr(thing, 'members') else [],
            member_of=[c.concept_id for c in thing.member_of],
            superordinates=[broader_concept.concept_id for broader_concept in thing.broader_concepts],
            infer_concept_relations=thing.infer_concept_relations
        )
    else:
        matches = {}
        for m in thing.matches:
            key = m.matchtype.name[:m.matchtype.name.find('Match')]
            if key not in matches:
                matches[key] = []
            matches[key].append(m.uri)
        return Concept(
            id=thing.concept_id,
            uri=thing.uri,
            concept_scheme=ConceptScheme(thing.conceptscheme.uri),
            labels=[
                Label(label.label, label.labeltype_id, label.language_id)
                for label in thing.labels
            ],
            notes=[
                Note(n.note, n.notetype_id, n.language_id)
                for n in thing.notes
            ],
            sources=[
                Source(s.citation)
                for s in thing.sources
            ],
            broader=[c.concept_id for c in thing.broader_concepts],
            narrower=[c.concept_id for c in thing.narrower_concepts],
            related=[c.concept_id for c in thing.related_concepts],
            member_of=[c.concept_id for c in thing.member_of],
            subordinate_arrays=[narrower_collection.concept_id for narrower_collection in thing.narrower_collections],
            matches=matches,
        )


def internal_providers_only(fn):
    """
    aspect oriented way to check if provider is internal when calling the decorated function

    :param fn: the decorated function
    :return: around advice
    :raises pyramid.httpexceptions.HTTPMethodNotAllowed: when provider is not internal
    """
    def advice(parent_object, *args, **kw):

        if isinstance(parent_object.provider, SQLAlchemyProvider) and \
                'external' not in parent_object.provider.get_metadata()['subject']:
            return fn(parent_object, *args, **kw)
        else:
            raise HTTPMethodNotAllowed()

    return advice


def update_last_visited_concepts(request, concept_data):
    deque_last_visited = deque(maxlen=4)
    deque_last_visited.extend(request.session.get('last_visited', []))
    try:
        # Try to remove concept from the queue to prevent double entries
        deque_last_visited.remove(concept_data)
    except ValueError:
        # Concept is not in the queue
        pass
    # Add concept to the queue
    deque_last_visited.append(concept_data)
    request.session['last_visited'] = list(deque_last_visited)


def label_sort(concepts, language='any'):
    if not concepts:
        return []
    return sorted(
        concepts, key=lambda concept: concept._sortkey(key='sortlabel',
                                                       language=language)
    )


def db_provider_to_skosprovider(db_provider: Provider) -> SQLAlchemyProvider:
    """Create a SQLAlchemyProvider from a atramhasis.data.models.Provider.

    :param db_provider: The Provider to use as basis for the SQLAlchemyProvider.
    :return: An SQLAlchemyProvider with the data from the `db_provider`
    """
    metadata = copy.deepcopy(db_provider.meta)
    metadata["conceptscheme_id"] = db_provider.conceptscheme_id
    metadata['atramhasis.id_generation_strategy'] = db_provider.id_generation_strategy
    metadata["id"] = db_provider.id
    return SQLAlchemyProvider(
        metadata=metadata,
        session=orm.object_session(db_provider),
        expand_strategy=db_provider.expand_strategy.value,
        uri_generator=UriPatternGenerator(db_provider.uri_pattern),
    )


1			"""
2			Module containing utility functions used by Atramhasis.
3			"""
4			import copy
5			from collections import deque
6
7			from pyramid.httpexceptions import HTTPMethodNotAllowed
8			from skosprovider.skos import Collection
9			from skosprovider.skos import Concept
10			from skosprovider.skos import ConceptScheme
11			from skosprovider.skos import Label
12			from skosprovider.skos import Note
13			from skosprovider.skos import Source
14			from skosprovider.uri import UriPatternGenerator
15			from skosprovider_sqlalchemy.providers import SQLAlchemyProvider
16			from sqlalchemy import orm
17			from sqlalchemy.orm import Session
18
19			from atramhasis.data.models import IDGenerationStrategy
20			from atramhasis.data.models import Provider
21
22
23			def from_thing(thing):
24			"""
25			Map a :class:`skosprovider_sqlalchemy.models.Thing` to a
26			:class:`skosprovider.skos.Concept` or
27			a :class:`skosprovider.skos.Collection`, depending on the type.
28
29			:param skosprovider_sqlalchemy.models.Thing thing: Thing to map.
30			:rtype: :class:`~skosprovider.skos.Concept` or
31			:class:`~skosprovider.skos.Collection`.
32			"""
33			if thing.type and thing.type == 'collection':
34			return Collection(
35			id=thing.concept_id,
36			uri=thing.uri,
37			concept_scheme=ConceptScheme(thing.conceptscheme.uri),
38			labels=[
39			Label(label.label, label.labeltype_id, label.language_id)
40			for label in thing.labels
41			],
42			notes=[
43			Note(n.note, n.notetype_id, n.language_id)
44			for n in thing.notes
45			],
46			sources=[
47			Source(s.citation)
48			for s in thing.sources
49			],
50			members=[member.concept_id for member in thing.members] if hasattr(thing, 'members') else [],
51			member_of=[c.concept_id for c in thing.member_of],
52			superordinates=[broader_concept.concept_id for broader_concept in thing.broader_concepts],
53			infer_concept_relations=thing.infer_concept_relations
54			)
55			else:
56			matches = {}
57			for m in thing.matches:
58			key = m.matchtype.name[:m.matchtype.name.find('Match')]
59			if key not in matches:
60			matches[key] = []
61			matches[key].append(m.uri)
62			return Concept(
63			id=thing.concept_id,
64			uri=thing.uri,
65			concept_scheme=ConceptScheme(thing.conceptscheme.uri),
66			labels=[
67			Label(label.label, label.labeltype_id, label.language_id)
68			for label in thing.labels
69			],
70			notes=[
71			Note(n.note, n.notetype_id, n.language_id)
72			for n in thing.notes
73			],
74			sources=[
75			Source(s.citation)
76			for s in thing.sources
77			],
78			broader=[c.concept_id for c in thing.broader_concepts],
79			narrower=[c.concept_id for c in thing.narrower_concepts],
80			related=[c.concept_id for c in thing.related_concepts],
81			member_of=[c.concept_id for c in thing.member_of],
82			subordinate_arrays=[narrower_collection.concept_id for narrower_collection in thing.narrower_collections],
83			matches=matches,
84			)
85
86
87			def internal_providers_only(fn):
88			"""
89			aspect oriented way to check if provider is internal when calling the decorated function
90
91			:param fn: the decorated function
92			:return: around advice
93			:raises pyramid.httpexceptions.HTTPMethodNotAllowed: when provider is not internal
94			"""
95			def advice(parent_object, args, *kw):
96
97			if isinstance(parent_object.provider, SQLAlchemyProvider) and \
98			'external' not in parent_object.provider.get_metadata()['subject']:
99			return fn(parent_object, args, *kw)
100			else:
101			raise HTTPMethodNotAllowed()
102
103			return advice
104
105
106			def update_last_visited_concepts(request, concept_data):
107			deque_last_visited = deque(maxlen=4)
108			deque_last_visited.extend(request.session.get('last_visited', []))
109			try:
110			# Try to remove concept from the queue to prevent double entries
111			deque_last_visited.remove(concept_data)
112			except ValueError:
113			# Concept is not in the queue
114			pass
115			# Add concept to the queue
116			deque_last_visited.append(concept_data)
117			request.session['last_visited'] = list(deque_last_visited)
118
119
120			def label_sort(concepts, language='any'):
121			if not concepts:
122			return []
123			return sorted(
124			concepts, key=lambda concept: concept._sortkey(key='sortlabel',
125			language=language)
126			)
127
128
129			def db_provider_to_skosprovider(db_provider: Provider) -> SQLAlchemyProvider:
130			"""Create a SQLAlchemyProvider from a atramhasis.data.models.Provider.
131
132			:param db_provider: The Provider to use as basis for the SQLAlchemyProvider.
133			:return: An SQLAlchemyProvider with the data from the `db_provider`
134			"""
135			metadata = copy.deepcopy(db_provider.meta)
136			metadata["conceptscheme_id"] = db_provider.conceptscheme_id
137			metadata['atramhasis.id_generation_strategy'] = db_provider.id_generation_strategy
138			metadata["id"] = db_provider.id
139			return SQLAlchemyProvider(
140			metadata=metadata,
141			session=orm.object_session(db_provider),
142			expand_strategy=db_provider.expand_strategy.value,
143			uri_generator=UriPatternGenerator(db_provider.uri_pattern),
144			)
145

OnroerendErfgoed / atramhasis

Pull Request — develop (#812)

atramhasis.utils.from_thing() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like