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

atramhasis.mappers.map_concept() F
last analyzed 2024-08-12 13:03 UTC

↳ Parent: atramhasis.mappers

Complexity

Conditions

Size

Total Lines	168
Code Lines	146

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	146
dl	0
loc	168
rs	0
c	0
b	0
f	0
cc	37
nop	3

How to fix Long Method Complexity

"""
Module containing mapping functions used by Atramhasis.
"""

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 Match
from skosprovider_sqlalchemy.models import Note
from skosprovider_sqlalchemy.models import Source
from skosprovider_sqlalchemy.providers import SQLAlchemyProvider
from sqlalchemy.exc import NoResultFound

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


def is_html(value):
    """
    Check if a value has html inside. Only tags checked <strong> <em> <a>.

    :param value: a string
    :return: a boolean (True, HTML present | False, no HTML present)
    """
    tag_list = ['<strong>', '<em>', '<a>', '</strong>', '</em>', '</a>', '<a']
    return any(tag in value for tag in tag_list)


def map_concept(concept, concept_json, skos_manager):
    """
    Map a concept from json to the database.

    :param skosprovider_sqlalchemy.models.Thing concept: A concept or
        collection as known to the database.
    :param dict concept_json: A dict representing the json sent to our REST
        service.
    :param skos_manager: A skos_manager to acces db operations
    :returns: The :class:`skosprovider_sqlalchemy.models.Thing` enhanced
        with the information from the json object.
    """
    concept_json_type = concept_json.get('type', None)
    if concept.type != concept_json_type:

        if concept_json_type == 'concept':
            members = concept.members
            concept = skos_manager.change_type(
                concept,
                concept.concept_id,
                concept.conceptscheme_id,
                concept_json_type,
                concept.uri
            )
            for member in members:
                if member.type == 'concept':
                    concept.narrower_concepts.add(member)
                elif member.type == 'collection':
                    concept.narrower_collections.add(member)
        elif concept_json_type == 'collection':
            narrower_concepts = concept.narrower_concepts
            narrower_collections = concept.narrower_collections
            concept = skos_manager.change_type(
                concept,
                concept.concept_id,
                concept.conceptscheme_id,
                concept_json_type,
                concept.uri
            )
            for narrower_concept in narrower_concepts:
                concept.members.add(narrower_concept)
            for narrower_collection in narrower_collections:
                concept.members.add(narrower_collection)
    elif concept_json_type == 'collection':
        concept.members.clear()
    elif concept_json_type == 'concept':
        concept.narrower_collections.clear()
        concept.narrower_concepts.clear()
    if concept.type in ('concept', 'collection'):
        concept.labels[:] = []
        labels = concept_json.get('labels', [])
        for label in labels:
            label = Label(
                label=label.get('label', ''),
                labeltype_id=label.get('type', ''),
                language_id=label.get('language', ''),
            )
            concept.labels.append(label)
        concept.notes[:] = []
        notes = concept_json.get('notes', [])
        for n in notes:
            note = Note(note=n.get('note', ''), notetype_id=n.get('type', ''), language_id=n.get('language', ''))
            if is_html(note.note):
                note.markup = 'HTML'
            concept.notes.append(note)
        concept.sources[:] = []
        sources = concept_json.get('sources', [])
        for s in sources:
            source = Source(citation=s.get('citation', ''))
            if is_html(source.citation):
                source.markup = 'HTML'
            concept.sources.append(source)

        concept.member_of.clear()
        member_of = concept_json.get('member_of', [])
        for memberof in member_of:
            try:
                memberof_collection = skos_manager.get_thing(
                    concept_id=memberof['id'],
                    conceptscheme_id=concept.conceptscheme_id)
            except NoResultFound:
                memberof_collection = Collection(concept_id=memberof['id'], conceptscheme_id=concept.conceptscheme_id)
            concept.member_of.add(memberof_collection)

        if concept.type == 'concept':
            concept.related_concepts.clear()
            related = concept_json.get('related', [])
            for related in related:
                try:
                    related_concept = skos_manager.get_thing(
                        concept_id=related['id'],
                        conceptscheme_id=concept.conceptscheme_id)
                except NoResultFound:
                    related_concept = Concept(concept_id=related['id'], conceptscheme_id=concept.conceptscheme_id)
                concept.related_concepts.add(related_concept)

            concept.broader_concepts.clear()
            broader = concept_json.get('broader', [])
            for broader in broader:
                try:
                    broader_concept = skos_manager.get_thing(
                        concept_id=broader['id'],
                        conceptscheme_id=concept.conceptscheme_id)
                except NoResultFound:
                    broader_concept = Concept(concept_id=broader['id'], conceptscheme_id=concept.conceptscheme_id)
                concept.broader_concepts.add(broader_concept)
            narrower = concept_json.get('narrower', [])
            for narrower in narrower:
                try:
                    narrower_concept = skos_manager.get_thing(
                        concept_id=narrower['id'],
                        conceptscheme_id=concept.conceptscheme_id)
                except NoResultFound:
                    narrower_concept = Concept(concept_id=narrower['id'], conceptscheme_id=concept.conceptscheme_id)
                concept.narrower_concepts.add(narrower_concept)

            matches = []
            matchdict = concept_json.get('matches', {})
            for match_type, uris in matchdict.items():
                db_type = match_type + "Match"
                match_type = skos_manager.get_match_type(db_type)
                for uri in uris:
                    concept_id = concept_json.get('id', -1)
                    try:
                        match = skos_manager.get_match(uri=uri, matchtype_id=match_type.name,
                                                       concept_id=concept_id)
                    except NoResultFound:
                        match = Match()
                        match.matchtype = match_type
                        match.uri = uri
                    matches.append(match)
            concept.matches = matches

            narrower_collections = concept_json.get('subordinate_arrays', [])
            for narrower in narrower_collections:
                try:
                    narrower_collection = skos_manager.get_thing(
                        concept_id=narrower['id'],
                        conceptscheme_id=concept.conceptscheme_id)
                except NoResultFound:
                    narrower_collection = Collection(concept_id=narrower['id'],
                                                     conceptscheme_id=concept.conceptscheme_id)
                concept.narrower_collections.add(narrower_collection)

        if concept.type == 'collection':
            members = concept_json.get('members', [])
            for member in members:
                try:
                    member_concept = skos_manager.get_thing(
                        concept_id=member['id'],
                        conceptscheme_id=concept.conceptscheme_id)
                except NoResultFound:
                    member_concept = Concept(concept_id=member['id'], conceptscheme_id=concept.conceptscheme_id)
                concept.members.add(member_concept)

            concept.broader_concepts.clear()
            broader_concepts = concept_json.get('superordinates', [])
            for broader in broader_concepts:
                try:
                    broader_concept = skos_manager.get_thing(
                        concept_id=broader['id'],
                        conceptscheme_id=concept.conceptscheme_id)
                except NoResultFound:
                    broader_concept = Concept(concept_id=broader['id'], conceptscheme_id=concept.conceptscheme_id)
                concept.broader_concepts.add(broader_concept)
            if 'infer_concept_relations' in concept_json:
                concept.infer_concept_relations = concept_json['infer_concept_relations']
    return concept


def map_conceptscheme(conceptscheme, conceptscheme_json):
    """
    Map a conceptscheme from json to the database.

    :param skosprovider_sqlalchemy.models.ConceptScheme conceptscheme: A conceptscheme as known to the database.
    :param dict conceptscheme_json: A dict representing the json sent to our REST
        service.
    :returns: The :class:`skosprovider_sqlalchemy.models.ConceptScheme` enhanced
        with the information from the json object.
    """
    conceptscheme.labels[:] = []
    labels = conceptscheme_json.get('labels', [])
    for label in labels:
        label = Label(
            label=label.get('label', ''),
            labeltype_id=label.get('type', ''),
            language_id=label.get('language', ''),
        )
        conceptscheme.labels.append(label)
    conceptscheme.notes[:] = []
    notes = conceptscheme_json.get('notes', [])
    for n in notes:
        note = Note(note=n.get('note', ''), notetype_id=n.get('type', ''), language_id=n.get('language', ''))
        conceptscheme.notes.append(note)
    conceptscheme.sources[:] = []
    sources = conceptscheme_json.get('sources', [])
    for s in sources:
        source = Source(citation=s.get('citation', ''))
        conceptscheme.sources.append(source)
    return conceptscheme


def map_provider(provider_json: dict, provider: Provider = None) -> Provider:
    """
    Create a atramhasis.data.models.Provider from json data.

    An existing provider can optionally be passed. When passed this one will be updated.

    :param provider_json: JSON data as a dict.
    :param provider: A provider which will be updated with the JSON data. When None
       a new Provider instance will be returned.
    :return: A Provider set with data from the JSON.
    """
    if provider is None:
        # Only executed on creation.
        provider = Provider()
        provider.conceptscheme = ConceptScheme(uri=provider_json['conceptscheme_uri'])
        provider.id = provider_json.get("id")

    provider.meta = provider_json.get("metadata") or {}
    provider.default_language = provider_json.get("default_language")
    provider.force_display_language = provider_json.get("force_display_language")
    provider.id_generation_strategy = IDGenerationStrategy[
        provider_json.get("id_generation_strategy") or 'NUMERIC'
    ]
    provider.subject = provider_json.get("subject") or []
    provider.uri_pattern = provider_json["uri_pattern"]
    provider.expand_strategy = ExpandStrategy[
        (provider_json.get("expand_strategy") or 'RECURSE').upper()
    ]
    return provider


1			"""
2			Module containing mapping functions used by Atramhasis.
3			"""
4
5			from skosprovider_sqlalchemy.models import Collection
6			from skosprovider_sqlalchemy.models import Concept
7			from skosprovider_sqlalchemy.models import ConceptScheme
8			from skosprovider_sqlalchemy.models import Label
9			from skosprovider_sqlalchemy.models import Match
10			from skosprovider_sqlalchemy.models import Note
11			from skosprovider_sqlalchemy.models import Source
12			from skosprovider_sqlalchemy.providers import SQLAlchemyProvider
13			from sqlalchemy.exc import NoResultFound
14
15			from atramhasis.data.models import ExpandStrategy
16			from atramhasis.data.models import IDGenerationStrategy
17			from atramhasis.data.models import Provider
18
19
20			def is_html(value):
21			"""
22			Check if a value has html inside. Only tags checked <strong> <em> <a>.
23
24			:param value: a string
25			:return: a boolean (True, HTML present \| False, no HTML present)
26			"""
27			tag_list = ['<strong>', '<em>', '<a>', '</strong>', '</em>', '</a>', '<a']
28			return any(tag in value for tag in tag_list)
29
30
31			def map_concept(concept, concept_json, skos_manager):
32			"""
33			Map a concept from json to the database.
34
35			:param skosprovider_sqlalchemy.models.Thing concept: A concept or
36			collection as known to the database.
37			:param dict concept_json: A dict representing the json sent to our REST
38			service.
39			:param skos_manager: A skos_manager to acces db operations
40			:returns: The :class:`skosprovider_sqlalchemy.models.Thing` enhanced
41			with the information from the json object.
42			"""
43			concept_json_type = concept_json.get('type', None)
44			if concept.type != concept_json_type:
45
46			if concept_json_type == 'concept':
47			members = concept.members
48			concept = skos_manager.change_type(
49			concept,
50			concept.concept_id,
51			concept.conceptscheme_id,
52			concept_json_type,
53			concept.uri
54			)
55			for member in members:
56			if member.type == 'concept':
57			concept.narrower_concepts.add(member)
58			elif member.type == 'collection':
59			concept.narrower_collections.add(member)
60			elif concept_json_type == 'collection':
61			narrower_concepts = concept.narrower_concepts
62			narrower_collections = concept.narrower_collections
63			concept = skos_manager.change_type(
64			concept,
65			concept.concept_id,
66			concept.conceptscheme_id,
67			concept_json_type,
68			concept.uri
69			)
70			for narrower_concept in narrower_concepts:
71			concept.members.add(narrower_concept)
72			for narrower_collection in narrower_collections:
73			concept.members.add(narrower_collection)
74			elif concept_json_type == 'collection':
75			concept.members.clear()
76			elif concept_json_type == 'concept':
77			concept.narrower_collections.clear()
78			concept.narrower_concepts.clear()
79			if concept.type in ('concept', 'collection'):
80			concept.labels[:] = []
81			labels = concept_json.get('labels', [])
82			for label in labels:
83			label = Label(
84			label=label.get('label', ''),
85			labeltype_id=label.get('type', ''),
86			language_id=label.get('language', ''),
87			)
88			concept.labels.append(label)
89			concept.notes[:] = []
90			notes = concept_json.get('notes', [])
91			for n in notes:
92			note = Note(note=n.get('note', ''), notetype_id=n.get('type', ''), language_id=n.get('language', ''))
93			if is_html(note.note):
94			note.markup = 'HTML'
95			concept.notes.append(note)
96			concept.sources[:] = []
97			sources = concept_json.get('sources', [])
98			for s in sources:
99			source = Source(citation=s.get('citation', ''))
100			if is_html(source.citation):
101			source.markup = 'HTML'
102			concept.sources.append(source)
103
104			concept.member_of.clear()
105			member_of = concept_json.get('member_of', [])
106			for memberof in member_of:
107			try:
108			memberof_collection = skos_manager.get_thing(
109			concept_id=memberof['id'],
110			conceptscheme_id=concept.conceptscheme_id)
111			except NoResultFound:
112			memberof_collection = Collection(concept_id=memberof['id'], conceptscheme_id=concept.conceptscheme_id)
113			concept.member_of.add(memberof_collection)
114
115			if concept.type == 'concept':
116			concept.related_concepts.clear()
117			related = concept_json.get('related', [])
118			for related in related:
119			try:
120			related_concept = skos_manager.get_thing(
121			concept_id=related['id'],
122			conceptscheme_id=concept.conceptscheme_id)
123			except NoResultFound:
124			related_concept = Concept(concept_id=related['id'], conceptscheme_id=concept.conceptscheme_id)
125			concept.related_concepts.add(related_concept)
126
127			concept.broader_concepts.clear()
128			broader = concept_json.get('broader', [])
129			for broader in broader:
130			try:
131			broader_concept = skos_manager.get_thing(
132			concept_id=broader['id'],
133			conceptscheme_id=concept.conceptscheme_id)
134			except NoResultFound:
135			broader_concept = Concept(concept_id=broader['id'], conceptscheme_id=concept.conceptscheme_id)
136			concept.broader_concepts.add(broader_concept)
137			narrower = concept_json.get('narrower', [])
138			for narrower in narrower:
139			try:
140			narrower_concept = skos_manager.get_thing(
141			concept_id=narrower['id'],
142			conceptscheme_id=concept.conceptscheme_id)
143			except NoResultFound:
144			narrower_concept = Concept(concept_id=narrower['id'], conceptscheme_id=concept.conceptscheme_id)
145			concept.narrower_concepts.add(narrower_concept)
146
147			matches = []
148			matchdict = concept_json.get('matches', {})
149			for match_type, uris in matchdict.items():
150			db_type = match_type + "Match"
151			match_type = skos_manager.get_match_type(db_type)
152			for uri in uris:
153			concept_id = concept_json.get('id', -1)
154			try:
155			match = skos_manager.get_match(uri=uri, matchtype_id=match_type.name,
156			concept_id=concept_id)
157			except NoResultFound:
158			match = Match()
159			match.matchtype = match_type
160			match.uri = uri
161			matches.append(match)
162			concept.matches = matches
163
164			narrower_collections = concept_json.get('subordinate_arrays', [])
165			for narrower in narrower_collections:
166			try:
167			narrower_collection = skos_manager.get_thing(
168			concept_id=narrower['id'],
169			conceptscheme_id=concept.conceptscheme_id)
170			except NoResultFound:
171			narrower_collection = Collection(concept_id=narrower['id'],
172			conceptscheme_id=concept.conceptscheme_id)
173			concept.narrower_collections.add(narrower_collection)
174
175			if concept.type == 'collection':
176			members = concept_json.get('members', [])
177			for member in members:
178			try:
179			member_concept = skos_manager.get_thing(
180			concept_id=member['id'],
181			conceptscheme_id=concept.conceptscheme_id)
182			except NoResultFound:
183			member_concept = Concept(concept_id=member['id'], conceptscheme_id=concept.conceptscheme_id)
184			concept.members.add(member_concept)
185
186			concept.broader_concepts.clear()
187			broader_concepts = concept_json.get('superordinates', [])
188			for broader in broader_concepts:
189			try:
190			broader_concept = skos_manager.get_thing(
191			concept_id=broader['id'],
192			conceptscheme_id=concept.conceptscheme_id)
193			except NoResultFound:
194			broader_concept = Concept(concept_id=broader['id'], conceptscheme_id=concept.conceptscheme_id)
195			concept.broader_concepts.add(broader_concept)
196			if 'infer_concept_relations' in concept_json:
197			concept.infer_concept_relations = concept_json['infer_concept_relations']
198			return concept
199
200
201			def map_conceptscheme(conceptscheme, conceptscheme_json):
202			"""
203			Map a conceptscheme from json to the database.
204
205			:param skosprovider_sqlalchemy.models.ConceptScheme conceptscheme: A conceptscheme as known to the database.
206			:param dict conceptscheme_json: A dict representing the json sent to our REST
207			service.
208			:returns: The :class:`skosprovider_sqlalchemy.models.ConceptScheme` enhanced
209			with the information from the json object.
210			"""
211			conceptscheme.labels[:] = []
212			labels = conceptscheme_json.get('labels', [])
213			for label in labels:
214			label = Label(
215			label=label.get('label', ''),
216			labeltype_id=label.get('type', ''),
217			language_id=label.get('language', ''),
218			)
219			conceptscheme.labels.append(label)
220			conceptscheme.notes[:] = []
221			notes = conceptscheme_json.get('notes', [])
222			for n in notes:
223			note = Note(note=n.get('note', ''), notetype_id=n.get('type', ''), language_id=n.get('language', ''))
224			conceptscheme.notes.append(note)
225			conceptscheme.sources[:] = []
226			sources = conceptscheme_json.get('sources', [])
227			for s in sources:
228			source = Source(citation=s.get('citation', ''))
229			conceptscheme.sources.append(source)
230			return conceptscheme
231
232
233			def map_provider(provider_json: dict, provider: Provider = None) -> Provider:
234			"""
235			Create a atramhasis.data.models.Provider from json data.
236
237			An existing provider can optionally be passed. When passed this one will be updated.
238
239			:param provider_json: JSON data as a dict.
240			:param provider: A provider which will be updated with the JSON data. When None
241			a new Provider instance will be returned.
242			:return: A Provider set with data from the JSON.
243			"""
244			if provider is None:
245			# Only executed on creation.
246			provider = Provider()
247			provider.conceptscheme = ConceptScheme(uri=provider_json['conceptscheme_uri'])
248			provider.id = provider_json.get("id")
249
250			provider.meta = provider_json.get("metadata") or {}
251			provider.default_language = provider_json.get("default_language")
252			provider.force_display_language = provider_json.get("force_display_language")
253			provider.id_generation_strategy = IDGenerationStrategy[
254			provider_json.get("id_generation_strategy") or 'NUMERIC'
255			]
256			provider.subject = provider_json.get("subject") or []
257			provider.uri_pattern = provider_json["uri_pattern"]
258			provider.expand_strategy = ExpandStrategy[
259			(provider_json.get("expand_strategy") or 'RECURSE').upper()
260			]
261			return provider
262

OnroerendErfgoed / atramhasis

atramhasis.mappers.map_concept() F last analyzed 2024-08-12 13:03 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

atramhasis.mappers.map_concept() F
last analyzed 2024-08-12 13:03 UTC