atramhasis.utils.from_thing() - Code Metrics - OnroerendErfgoed/atramhasis - Measure and Improve Code Quality continuously with Scrutinizer

atramhasis.utils.from_thing() B
last analyzed 2024-08-12 13:03 UTC

↳ 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 contextlib
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 engine_from_config
from sqlalchemy import orm
from sqlalchemy.orm import sessionmaker

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),
    )


@contextlib.contextmanager
def db_session(settings):
    engine = engine_from_config(settings, 'sqlalchemy.')
    session_maker = sessionmaker(bind=engine)
    session = session_maker()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()


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

OnroerendErfgoed / atramhasis

atramhasis.utils.from_thing() B last analyzed 2024-08-12 13:03 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

atramhasis.utils.from_thing() B
last analyzed 2024-08-12 13:03 UTC