atramhasis.renderers.CSVRenderer.__init__() - Code Metrics - OnroerendErfgoed/atramhasis - Measure and Improve Code Quality continuously with Scrutinizer

atramhasis.renderers.CSVRenderer.init() A
last analyzed 2024-08-12 13:03 UTC

↳ Parent: atramhasis.renderers

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

import copy
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.data.models 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, *_):
    if provider.uri_generator:
        uri_pattern = provider.uri_generator.pattern
    else:
        uri_pattern = None

    metadata = copy.deepcopy(provider.metadata)
    metadata.pop('id', None)
    metadata.pop('conceptscheme_id', None)
    data = {
        'id': provider.metadata['id'],
        'type': provider.__class__.__name__,
        'conceptscheme_uri': provider.concept_scheme.uri,
        'uri_pattern': uri_pattern,
        'default_language': metadata.pop("default_language", None),
        'subject': metadata.pop("subject", None),
        'force_display_language': metadata.pop("atramhasis.force_display_language", None),
        'metadata': metadata,
    }
    return data


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

OnroerendErfgoed / atramhasis

atramhasis.renderers.CSVRenderer.__init__() A last analyzed 2024-08-12 13:03 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

atramhasis.renderers.CSVRenderer.init() A
last analyzed 2024-08-12 13:03 UTC