Test Failed
Push — master ( 9ea4cd...0ffab9 )
by Koen
03:34 queued 13s
created

atramhasis.mappers.map_concept()   F

Complexity

Conditions 37

Size

Total Lines 164
Code Lines 143

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 143
dl 0
loc 164
rs 0
c 0
b 0
f 0
cc 37
nop 3

How to fix   Long Method    Complexity   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

Complexity

Complex classes like atramhasis.mappers.map_concept() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

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 Label
8
from skosprovider_sqlalchemy.models import Match
9
from skosprovider_sqlalchemy.models import Note
10
from skosprovider_sqlalchemy.models import Source
11
from sqlalchemy.orm.exc import NoResultFound
12
13
14
def is_html(value):
15
    """
16
    Check if a value has html inside. Only tags checked <strong> <em> <a>.
17
18
    :param value: a string
19
    :return: a boolean (True, HTML present | False, no HTML present)
20
    """
21
    tag_list = ['<strong>', '<em>', '<a>', '</strong>', '</em>', '</a>', '<a']
22
    return any(tag in value for tag in tag_list)
23
24
25
def map_concept(concept, concept_json, skos_manager):
26
    """
27
    Map a concept from json to the database.
28
29
    :param skosprovider_sqlalchemy.models.Thing concept: A concept or
30
        collection as known to the database.
31
    :param dict concept_json: A dict representing the json sent to our REST
32
        service.
33
    :param skos_manager: A skos_manager to acces db operations
34
    :returns: The :class:`skosprovider_sqlalchemy.models.Thing` enhanced
35
        with the information from the json object.
36
    """
37
    concept_json_type = concept_json.get('type', None)
38
    if concept.type != concept_json_type:
39
40
        if concept_json_type == 'concept':
41
            members = concept.members
42
            concept = skos_manager.change_type(
43
                concept,
44
                concept.concept_id,
45
                concept.conceptscheme_id,
46
                concept_json_type,
47
                concept.uri
48
            )
49
            for member in members:
50
                if member.type == 'concept':
51
                    concept.narrower_concepts.add(member)
52
                elif member.type == 'collection':
53
                    concept.narrower_collections.add(member)
54
        elif concept_json_type == 'collection':
55
            narrower_concepts = concept.narrower_concepts
56
            narrower_collections = concept.narrower_collections
57
            concept = skos_manager.change_type(
58
                concept,
59
                concept.concept_id,
60
                concept.conceptscheme_id,
61
                concept_json_type,
62
                concept.uri
63
            )
64
            for narrower_concept in narrower_concepts:
65
                concept.members.add(narrower_concept)
66
            for narrower_collection in narrower_collections:
67
                concept.members.add(narrower_collection)
68
    elif concept_json_type == 'collection':
69
        concept.members.clear()
70
    elif concept_json_type == 'concept':
71
        concept.narrower_collections.clear()
72
        concept.narrower_concepts.clear()
73
    if concept.type in ('concept', 'collection'):
74
        concept.labels[:] = []
75
        labels = concept_json.get('labels', [])
76
        for l in labels:
77
            label = Label(label=l.get('label', ''), labeltype_id=l.get('type', ''), language_id=l.get('language', ''))
78
            concept.labels.append(label)
79
        concept.notes[:] = []
80
        notes = concept_json.get('notes', [])
81
        for n in notes:
82
            note = Note(note=n.get('note', ''), notetype_id=n.get('type', ''), language_id=n.get('language', ''))
83
            if is_html(note.note):
84
                note.markup = 'HTML'
85
            concept.notes.append(note)
86
        concept.sources[:] = []
87
        sources = concept_json.get('sources', [])
88
        for s in sources:
89
            source = Source(citation=s.get('citation', ''))
90
            if is_html(source.citation):
91
                source.markup = 'HTML'
92
            concept.sources.append(source)
93
94
        concept.member_of.clear()
95
        member_of = concept_json.get('member_of', [])
96
        for memberof in member_of:
97
            try:
98
                memberof_collection = skos_manager.get_thing(
99
                    concept_id=memberof['id'],
100
                    conceptscheme_id=concept.conceptscheme_id)
101
            except NoResultFound:
102
                memberof_collection = Collection(concept_id=memberof['id'], conceptscheme_id=concept.conceptscheme_id)
103
            concept.member_of.add(memberof_collection)
104
105
        if concept.type == 'concept':
106
            concept.related_concepts.clear()
107
            related = concept_json.get('related', [])
108
            for related in related:
109
                try:
110
                    related_concept = skos_manager.get_thing(
111
                        concept_id=related['id'],
112
                        conceptscheme_id=concept.conceptscheme_id)
113
                except NoResultFound:
114
                    related_concept = Concept(concept_id=related['id'], conceptscheme_id=concept.conceptscheme_id)
115
                concept.related_concepts.add(related_concept)
116
117
            concept.broader_concepts.clear()
118
            broader = concept_json.get('broader', [])
119
            for broader in broader:
120
                try:
121
                    broader_concept = skos_manager.get_thing(
122
                        concept_id=broader['id'],
123
                        conceptscheme_id=concept.conceptscheme_id)
124
                except NoResultFound:
125
                    broader_concept = Concept(concept_id=broader['id'], conceptscheme_id=concept.conceptscheme_id)
126
                concept.broader_concepts.add(broader_concept)
127
            narrower = concept_json.get('narrower', [])
128
            for narrower in narrower:
129
                try:
130
                    narrower_concept = skos_manager.get_thing(
131
                        concept_id=narrower['id'],
132
                        conceptscheme_id=concept.conceptscheme_id)
133
                except NoResultFound:
134
                    narrower_concept = Concept(concept_id=narrower['id'], conceptscheme_id=concept.conceptscheme_id)
135
                concept.narrower_concepts.add(narrower_concept)
136
137
            matches = []
138
            matchdict = concept_json.get('matches', {})
139
            for type in matchdict:
140
                db_type = type + "Match"
141
                matchtype = skos_manager.get_match_type(db_type)
142
                for uri in matchdict[type]:
143
                    concept_id = concept_json.get('id', -1)
144
                    try:
145
                        match = skos_manager.get_match(uri=uri, matchtype_id=matchtype.name,
146
                                                       concept_id=concept_id)
147
                    except NoResultFound:
148
                        match = Match()
149
                        match.matchtype = matchtype
150
                        match.uri = uri
151
                    matches.append(match)
152
            concept.matches = matches
153
154
            narrower_collections = concept_json.get('subordinate_arrays', [])
155
            for narrower in narrower_collections:
156
                try:
157
                    narrower_collection = skos_manager.get_thing(
158
                        concept_id=narrower['id'],
159
                        conceptscheme_id=concept.conceptscheme_id)
160
                except NoResultFound:
161
                    narrower_collection = Collection(concept_id=narrower['id'],
162
                                                     conceptscheme_id=concept.conceptscheme_id)
163
                concept.narrower_collections.add(narrower_collection)
164
165
        if concept.type == 'collection':
166
            members = concept_json.get('members', [])
167
            for member in members:
168
                try:
169
                    member_concept = skos_manager.get_thing(
170
                        concept_id=member['id'],
171
                        conceptscheme_id=concept.conceptscheme_id)
172
                except NoResultFound:
173
                    member_concept = Concept(concept_id=member['id'], conceptscheme_id=concept.conceptscheme_id)
174
                concept.members.add(member_concept)
175
176
            concept.broader_concepts.clear()
177
            broader_concepts = concept_json.get('superordinates', [])
178
            for broader in broader_concepts:
179
                try:
180
                    broader_concept = skos_manager.get_thing(
181
                        concept_id=broader['id'],
182
                        conceptscheme_id=concept.conceptscheme_id)
183
                except NoResultFound:
184
                    broader_concept = Concept(concept_id=broader['id'], conceptscheme_id=concept.conceptscheme_id)
185
                concept.broader_concepts.add(broader_concept)
186
            if 'infer_concept_relations' in concept_json:
187
                concept.infer_concept_relations = concept_json['infer_concept_relations']
188
    return concept
189
190
191
def map_conceptscheme(conceptscheme, conceptscheme_json):
192
    """
193
    Map a conceptscheme from json to the database.
194
195
    :param skosprovider_sqlalchemy.models.ConceptScheme conceptscheme: A conceptscheme as known to the database.
196
    :param dict conceptscheme_json: A dict representing the json sent to our REST
197
        service.
198
    :returns: The :class:`skosprovider_sqlalchemy.models.ConceptScheme` enhanced
199
        with the information from the json object.
200
    """
201
    conceptscheme.labels[:] = []
202
    labels = conceptscheme_json.get('labels', [])
203
    for l in labels:
204
        label = Label(label=l.get('label', ''), labeltype_id=l.get('type', ''), language_id=l.get('language', ''))
205
        conceptscheme.labels.append(label)
206
    conceptscheme.notes[:] = []
207
    notes = conceptscheme_json.get('notes', [])
208
    for n in notes:
209
        note = Note(note=n.get('note', ''), notetype_id=n.get('type', ''), language_id=n.get('language', ''))
210
        conceptscheme.notes.append(note)
211
    conceptscheme.sources[:] = []
212
    sources = conceptscheme_json.get('sources', [])
213
    for s in sources:
214
        source = Source(citation=s.get('citation', ''))
215
        conceptscheme.sources.append(source)
216
    return conceptscheme
217