atramhasis.renderers.sa_provider_adapter() - Code Metrics - Inspection of "Add GET provider endpoints. (#795)" - OnroerendErfgoed/atramhasis - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( da51e8...70dbf1 )

unknown

created 2023-03-09 12:38 UTC

atramhasis.renderers.sa_provider_adapter() A

↳ Parent: atramhasis.renderers

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	2
dl	0
loc	7
rs	10
c	0
b	0
f	0

import csv
from io import StringIO

from pyramid.renderers import JSON
from pyramid_skosprovider.renderers import collection_adapter as skos_collection_adapter
from pyramid_skosprovider.renderers import concept_adapter as skos_concept_adapter
from pyramid_skosprovider.renderers import label_adapter as skos_label_adapter
from pyramid_skosprovider.renderers import note_adapter as skos_note_adapter
from pyramid_skosprovider.renderers import source_adapter as skos_source_adapter
from skosprovider.providers import VocabularyProvider
from skosprovider.skos import Collection as SkosCollection
from skosprovider.skos import Concept as SkosConcept
from skosprovider.skos import Label as SkosLabel
from skosprovider.skos import Note as SkosNote
from skosprovider.skos import Source as SkosSource
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 Language
from skosprovider_sqlalchemy.models import Note
from skosprovider_sqlalchemy.models import Source
from skosprovider_sqlalchemy.providers import SQLAlchemyProvider

from atramhasis.skos import IDGenerationStrategy


class CSVRenderer:
    def __init__(self, info):
        pass

    def __call__(self, value, system):
        f_out = StringIO()
        writer = csv.writer(f_out, delimiter=',', quoting=csv.QUOTE_ALL)

        writer.writerow(value['header'])
        writer.writerows(value['rows'])

        resp = system['request'].response
        resp.content_type = 'text/csv'
        resp.content_disposition = 'attachment;filename="' + value['filename'] + '.csv"'
        return f_out.getvalue()


json_renderer_verbose = JSON()


def conceptscheme_adapter(obj, request):
    """
    Adapter for rendering a :class:`skosprovider_sqlalchemy.models.ConceptScheme` to json

    :param skosprovider_sqlalchemy.models.ConceptScheme obj: The conceptscheme to be rendered.
    :param request: the current request
    :rtype: :class:`dict`
    """
    scheme_id = request.matchdict['scheme_id']
    provider = request.skos_registry.get_provider(scheme_id)
    language = request.params.get('language', request.locale_name)
    label = obj.label(language)
    return {
        'id': obj.id,
        'uri': obj.uri,
        'label': label.label if label else None,
        'subject': provider.metadata['subject'] if provider.metadata['subject'] else [],
        'labels': obj.labels,
        'notes': obj.notes,
        'languages': obj.languages,
        'sources': obj.sources
    }


def concept_adapter(obj, request):
    """
    Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Concept` to json with verbose relations.

    :param skosprovider_sqlalchemy.models.Concept obj: The concept to be rendered.
    :param request: the current request
    :rtype: :class:`dict`
    """
    matches = {}
    for m in obj.matches:
        key = m.matchtype.name[:m.matchtype.name.find('Match')]
        if key not in matches:
            matches[key] = []
        matches[key].append(m.uri)
    language = request.params.get('language', request.locale_name)
    label = obj.label(language)
    return {
        'id': obj.concept_id,
        'type': obj.type,
        'uri': obj.uri,
        'label': label.label if label else None,
        'labels': obj.labels,
        'notes': obj.notes,
        'sources': obj.sources,
        'broader': [map_relation(c, language) for c in obj.broader_concepts],
        'narrower': [map_relation(c, language) for c in obj.narrower_concepts],
        'related': [map_relation(c, language) for c in obj.related_concepts],
        'member_of': [map_relation(c, language) for c in obj.member_of],
        'subordinate_arrays': [map_relation(c, language) for c in obj.narrower_collections],
        'matches': matches
    }


def collection_adapter(obj, request):
    """
    Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Collection` to json with verbose relations.

    :param skosprovider_sqlalchemy.models.Collection obj: The collection to be rendered.
    :param request: the current request
    :rtype: :class:`dict`
    """
    language = request.params.get('language', request.locale_name)
    label = obj.label(language)
    return {
        'id': obj.concept_id,
        'type': obj.type,
        'uri': obj.uri,
        'label': label.label if label else None,
        'notes': obj.notes,
        'labels': obj.labels,
        'sources': obj.sources,
        'members': [map_relation(c, language) for c in obj.members],
        'member_of': [map_relation(c, language) for c in obj.member_of],
        'superordinates': [map_relation(c, language) for c in obj.broader_concepts],
        'infer_concept_relations': obj.infer_concept_relations
    }


def map_relation(thing, language='any'):
    """
    Map thing in a relation, leaving out the relations (to avoid circular dependencies)
    :param thing: the thing to map
    :param language: language to use for the label
    :rtype: :class:`dict`
    """
    label = thing.label(language)
    return {
        'id': thing.concept_id,
        'type': thing.type,
        'uri': thing.uri,
        'label': label.label if label else None
    }


def label_adapter(obj, request):
    """
    Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Label` to json.

    :param skosprovider_sqlalchemy.models.Label obj: The label to be rendered.
    :param request: the current request
    :rtype: :class:`dict`
    """
    return {
        'label': obj.label,
        'type': obj.labeltype_id,
        'language': obj.language_id
    }


def note_adapter(obj, request):
    """
    Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Note` to json.

    :param skosprovider_sqlalchemy.models.Note obj: The note to be rendered.
    :param request: the current request
    :rtype: :class:`dict`
    """
    return {
        'note': obj.note,
        'type': obj.notetype_id,
        'language': obj.language_id
    }


def source_adapter(obj, request):
    """
    Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Source` to json.

    :param skosprovider_sqlalchemy.models.Source obj: The source to be rendered.
    :param request: the current request
    :rtype: :class:`dict`
    """
    return {
        'citation': obj.citation
    }


def language_adaptor(obj, request):
    """
    Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Language` to json.

    :param skosprovider_sqlalchemy.models.Language obj: The language to be rendered.
    :param request: the current request
    :rtype: :class:`dict`
    """
    return {
        'id': obj.id,
        'name': obj.name
    }


def provider_adapter(provider: VocabularyProvider, *_):
    data = {
       'id': provider.metadata['id'],
       'type': provider.__class__.__name__,
       'conceptscheme_uri': provider.concept_scheme.uri,
       'uri_pattern': provider.uri_generator.pattern,
       'default_language': provider.metadata.get("default_language"),
       'subject': provider.metadata.get("subject"),
       'force_display_language': provider.metadata.get("atramhasis.force_display_language"),
    }
    return data


def sa_provider_adapter(provider: SQLAlchemyProvider, *args):
    data = provider_adapter(provider, *args)
    strategy = provider.metadata.get(
       "atramhasis.id_generation_strategy", IDGenerationStrategy.NUMERIC
    )
    data['id_generation_strategy'] = strategy.name
    return data


json_renderer_verbose.add_adapter(ConceptScheme, conceptscheme_adapter)
json_renderer_verbose.add_adapter(Concept, concept_adapter)
json_renderer_verbose.add_adapter(Collection, collection_adapter)
json_renderer_verbose.add_adapter(Label, label_adapter)
json_renderer_verbose.add_adapter(Note, note_adapter)
json_renderer_verbose.add_adapter(Source, source_adapter)
json_renderer_verbose.add_adapter(Language, language_adaptor)
json_renderer_verbose.add_adapter(SkosConcept, skos_concept_adapter)
json_renderer_verbose.add_adapter(SkosCollection, skos_collection_adapter)
json_renderer_verbose.add_adapter(SkosLabel, skos_label_adapter)
json_renderer_verbose.add_adapter(SkosNote, skos_note_adapter)
json_renderer_verbose.add_adapter(SkosSource, skos_source_adapter)
json_renderer_verbose.add_adapter(VocabularyProvider, provider_adapter)
json_renderer_verbose.add_adapter(SQLAlchemyProvider, sa_provider_adapter)


1			import csv
2			from io import StringIO
3
4			from pyramid.renderers import JSON
5			from pyramid_skosprovider.renderers import collection_adapter as skos_collection_adapter
6			from pyramid_skosprovider.renderers import concept_adapter as skos_concept_adapter
7			from pyramid_skosprovider.renderers import label_adapter as skos_label_adapter
8			from pyramid_skosprovider.renderers import note_adapter as skos_note_adapter
9			from pyramid_skosprovider.renderers import source_adapter as skos_source_adapter
10			from skosprovider.providers import VocabularyProvider
11			from skosprovider.skos import Collection as SkosCollection
12			from skosprovider.skos import Concept as SkosConcept
13			from skosprovider.skos import Label as SkosLabel
14			from skosprovider.skos import Note as SkosNote
15			from skosprovider.skos import Source as SkosSource
16			from skosprovider_sqlalchemy.models import Collection
17			from skosprovider_sqlalchemy.models import Concept
18			from skosprovider_sqlalchemy.models import ConceptScheme
19			from skosprovider_sqlalchemy.models import Label
20			from skosprovider_sqlalchemy.models import Language
21			from skosprovider_sqlalchemy.models import Note
22			from skosprovider_sqlalchemy.models import Source
23			from skosprovider_sqlalchemy.providers import SQLAlchemyProvider
24
25			from atramhasis.skos import IDGenerationStrategy
26
27
28			class CSVRenderer:
29			def __init__(self, info):
30			pass
31
32			def __call__(self, value, system):
33			f_out = StringIO()
34			writer = csv.writer(f_out, delimiter=',', quoting=csv.QUOTE_ALL)
35
36			writer.writerow(value['header'])
37			writer.writerows(value['rows'])
38
39			resp = system['request'].response
40			resp.content_type = 'text/csv'
41			resp.content_disposition = 'attachment;filename="' + value['filename'] + '.csv"'
42			return f_out.getvalue()
43
44
45			json_renderer_verbose = JSON()
46
47
48			def conceptscheme_adapter(obj, request):
49			"""
50			Adapter for rendering a :class:`skosprovider_sqlalchemy.models.ConceptScheme` to json
51
52			:param skosprovider_sqlalchemy.models.ConceptScheme obj: The conceptscheme to be rendered.
53			:param request: the current request
54			:rtype: :class:`dict`
55			"""
56			scheme_id = request.matchdict['scheme_id']
57			provider = request.skos_registry.get_provider(scheme_id)
58			language = request.params.get('language', request.locale_name)
59			label = obj.label(language)
60			return {
61			'id': obj.id,
62			'uri': obj.uri,
63			'label': label.label if label else None,
64			'subject': provider.metadata['subject'] if provider.metadata['subject'] else [],
65			'labels': obj.labels,
66			'notes': obj.notes,
67			'languages': obj.languages,
68			'sources': obj.sources
69			}
70
71
72			def concept_adapter(obj, request):
73			"""
74			Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Concept` to json with verbose relations.
75
76			:param skosprovider_sqlalchemy.models.Concept obj: The concept to be rendered.
77			:param request: the current request
78			:rtype: :class:`dict`
79			"""
80			matches = {}
81			for m in obj.matches:
82			key = m.matchtype.name[:m.matchtype.name.find('Match')]
83			if key not in matches:
84			matches[key] = []
85			matches[key].append(m.uri)
86			language = request.params.get('language', request.locale_name)
87			label = obj.label(language)
88			return {
89			'id': obj.concept_id,
90			'type': obj.type,
91			'uri': obj.uri,
92			'label': label.label if label else None,
93			'labels': obj.labels,
94			'notes': obj.notes,
95			'sources': obj.sources,
96			'broader': [map_relation(c, language) for c in obj.broader_concepts],
97			'narrower': [map_relation(c, language) for c in obj.narrower_concepts],
98			'related': [map_relation(c, language) for c in obj.related_concepts],
99			'member_of': [map_relation(c, language) for c in obj.member_of],
100			'subordinate_arrays': [map_relation(c, language) for c in obj.narrower_collections],
101			'matches': matches
102			}
103
104
105			def collection_adapter(obj, request):
106			"""
107			Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Collection` to json with verbose relations.
108
109			:param skosprovider_sqlalchemy.models.Collection obj: The collection to be rendered.
110			:param request: the current request
111			:rtype: :class:`dict`
112			"""
113			language = request.params.get('language', request.locale_name)
114			label = obj.label(language)
115			return {
116			'id': obj.concept_id,
117			'type': obj.type,
118			'uri': obj.uri,
119			'label': label.label if label else None,
120			'notes': obj.notes,
121			'labels': obj.labels,
122			'sources': obj.sources,
123			'members': [map_relation(c, language) for c in obj.members],
124			'member_of': [map_relation(c, language) for c in obj.member_of],
125			'superordinates': [map_relation(c, language) for c in obj.broader_concepts],
126			'infer_concept_relations': obj.infer_concept_relations
127			}
128
129
130			def map_relation(thing, language='any'):
131			"""
132			Map thing in a relation, leaving out the relations (to avoid circular dependencies)
133			:param thing: the thing to map
134			:param language: language to use for the label
135			:rtype: :class:`dict`
136			"""
137			label = thing.label(language)
138			return {
139			'id': thing.concept_id,
140			'type': thing.type,
141			'uri': thing.uri,
142			'label': label.label if label else None
143			}
144
145
146			def label_adapter(obj, request):
147			"""
148			Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Label` to json.
149
150			:param skosprovider_sqlalchemy.models.Label obj: The label to be rendered.
151			:param request: the current request
152			:rtype: :class:`dict`
153			"""
154			return {
155			'label': obj.label,
156			'type': obj.labeltype_id,
157			'language': obj.language_id
158			}
159
160
161			def note_adapter(obj, request):
162			"""
163			Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Note` to json.
164
165			:param skosprovider_sqlalchemy.models.Note obj: The note to be rendered.
166			:param request: the current request
167			:rtype: :class:`dict`
168			"""
169			return {
170			'note': obj.note,
171			'type': obj.notetype_id,
172			'language': obj.language_id
173			}
174
175
176			def source_adapter(obj, request):
177			"""
178			Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Source` to json.
179
180			:param skosprovider_sqlalchemy.models.Source obj: The source to be rendered.
181			:param request: the current request
182			:rtype: :class:`dict`
183			"""
184			return {
185			'citation': obj.citation
186			}
187
188
189			def language_adaptor(obj, request):
190			"""
191			Adapter for rendering a :class:`skosprovider_sqlalchemy.models.Language` to json.
192
193			:param skosprovider_sqlalchemy.models.Language obj: The language to be rendered.
194			:param request: the current request
195			:rtype: :class:`dict`
196			"""
197			return {
198			'id': obj.id,
199			'name': obj.name
200			}
201
202
203			def provider_adapter(provider: VocabularyProvider, *_):
204			data = {
205			'id': provider.metadata['id'],
206			'type': provider.__class__.__name__,
207			'conceptscheme_uri': provider.concept_scheme.uri,
208			'uri_pattern': provider.uri_generator.pattern,
209			'default_language': provider.metadata.get("default_language"),
210			'subject': provider.metadata.get("subject"),
211			'force_display_language': provider.metadata.get("atramhasis.force_display_language"),
212			}
213			return data
214
215
216			def sa_provider_adapter(provider: SQLAlchemyProvider, *args):
217			data = provider_adapter(provider, *args)
218			strategy = provider.metadata.get(
219			"atramhasis.id_generation_strategy", IDGenerationStrategy.NUMERIC
220			)
221			data['id_generation_strategy'] = strategy.name
222			return data
223
224
225			json_renderer_verbose.add_adapter(ConceptScheme, conceptscheme_adapter)
226			json_renderer_verbose.add_adapter(Concept, concept_adapter)
227			json_renderer_verbose.add_adapter(Collection, collection_adapter)
228			json_renderer_verbose.add_adapter(Label, label_adapter)
229			json_renderer_verbose.add_adapter(Note, note_adapter)
230			json_renderer_verbose.add_adapter(Source, source_adapter)
231			json_renderer_verbose.add_adapter(Language, language_adaptor)
232			json_renderer_verbose.add_adapter(SkosConcept, skos_concept_adapter)
233			json_renderer_verbose.add_adapter(SkosCollection, skos_collection_adapter)
234			json_renderer_verbose.add_adapter(SkosLabel, skos_label_adapter)
235			json_renderer_verbose.add_adapter(SkosNote, skos_note_adapter)
236			json_renderer_verbose.add_adapter(SkosSource, skos_source_adapter)
237			json_renderer_verbose.add_adapter(VocabularyProvider, provider_adapter)
238			json_renderer_verbose.add_adapter(SQLAlchemyProvider, sa_provider_adapter)
239

OnroerendErfgoed / atramhasis

Push — develop ( da51e8...70dbf1 )

atramhasis.renderers.sa_provider_adapter() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like