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

atramhasis.utils.from_thing() B
last analyzed 2025-11-28 09:40 UTC

↳ Parent: atramhasis.utils

Complexity

Conditions

Size

Total Lines	56
Code Lines	41

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
eloc	41
nop	1
dl	0
loc	56
rs	7.9626
c	0
b	0
f	0

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

OnroerendErfgoed / atramhasis

atramhasis.utils.from_thing() B last analyzed 2025-11-28 09:40 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 2025-11-28 09:40 UTC