Registry.get_provider() - Code Metrics - Inspection of "Update based on scrutinizer feedback." - OnroerendErfgoed/skosprovider - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — master (d5006c)

by Koen

created 2016-04-19 09:12 UTC

Registry.get_provider() A

↳ Parent: Registry

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	3
dl	0
loc	15
rs	9.4285

# -*- coding: utf-8 -*-

'''This module provides a registry for skos providers.

This registry helps us find providers during runtime. We can also apply some
operations to all or several providers at the same time.
'''

from __future__ import unicode_literals


class RegistryException(Exception):
    pass


class Registry:
    '''
    This registry collects all skos providers.
    '''

    providers = {}
    '''
    Dictionary containing all providers, keyed by id.
    '''

    concept_scheme_uri_map = {}
    '''
    Dictionary mapping concept scheme uri's to vocabulary id's.
    '''

    def __init__(self):
        self.providers = {}
        self.concept_scheme_uri_map = {}

    def register_provider(self, provider):
        '''
        Register a :class:`skosprovider.providers.VocabularyProvider`.

        :param skosprovider.providers.VocabularyProvider provider: The provider
            to register.
        '''
        if provider.get_vocabulary_id() in self.providers:
            raise RegistryException(
                'A provider with this id has already been registered.')
        self.providers[provider.get_vocabulary_id()] = provider
        self.concept_scheme_uri_map[provider.concept_scheme.uri] = provider.get_vocabulary_id()

    def remove_provider(self, id):
        '''
        Remove the provider with the given id or :term:`URI`.

        :param str id: The identifier for the provider.
        :returns: A :class:`skosprovider.providers.VocabularyProvider` or
            `False` if the id is unknown.
        '''
        if id in self.providers:
            p = self.providers.get(id, False)
            del self.providers[id]
            del self.concept_scheme_uri_map[p.concept_scheme.uri]
            return p
        elif id in self.concept_scheme_uri_map:
            id = self.concept_scheme_uri_map[id]
            return self.remove_provider(id)
        else:
            return False

    def get_provider(self, id):
        '''
        Get a provider by id or :term:`uri`.

        :param str id: The identifier for the provider. This can either be the
            id with which it was registered or the :term:`uri` of the conceptscheme
            that the provider services.
        :returns: A :class:`skosprovider.providers.VocabularyProvider`
            or `False` if the id or uri is unknown.
        '''
        if id in self.providers:
            return self.providers.get(id, False)
        elif id in self.concept_scheme_uri_map:
            return self.providers.get(self.concept_scheme_uri_map[id], False)
        return False

    def get_providers(self, **kwargs):
        '''Get all providers registered.

        If keyword `ids` is present, get only the providers with these ids.

        If keys `subject` is present, get only the providers that have this subject.

        .. code-block:: python

           # Get all providers with subject 'biology'
           registry.get_providers(subject='biology')

           # Get all providers with id 1 or 2
           registry.get_providers(ids=[1,2])

           # Get all providers with id 1 or 2 and subject 'biology'
           registry.get_providers(ids=[1,2], subject='biology']

        :param list ids: Only return providers with one of the Ids or uris.
        :param str subject: Only return providers with this subject.
        :returns: A list of :class:`providers <skosprovider.providers.VocabularyProvider>`
        '''
        if 'ids' in kwargs:
            ids = [self.concept_scheme_uri_map.get(id, id) for id in kwargs['ids']]
            providers = [
                self.providers[k] for k in self.providers.keys() if k in ids
            ]
        else:
            providers = list(self.providers.values())
        if 'subject' in kwargs:
            providers = [p for p in providers if kwargs['subject'] in p.metadata['subject']]
        return providers

    def find(self, query, **kwargs):
        '''Launch a query across all or a selection of providers.

        .. code-block:: python

            # Find anything that has a label of church in any provider.
            registry.find({'label': 'church'})

            # Find anything that has a label of church with the BUILDINGS provider.
            # Attention, this syntax was deprecated in version 0.3.0
            registry.find({'label': 'church'}, providers=['BUILDINGS'])

            # Find anything that has a label of church with the BUILDINGS provider.
            registry.find({'label': 'church'}, providers={'ids': ['BUILDINGS']})

            # Find anything that has a label of church with a provider
            # marked with the subject 'architecture'.
            registry.find({'label': 'church'}, providers={'subject': 'architecture'})

            # Find anything that has a label of church in any provider.
            # If possible, display the results with a Dutch label.
            registry.find({'label': 'church'}, language='nl')

        :param dict query: The query parameters that will be passed on to each
            :meth:`~skosprovider.providers.VocabularyProvider.find` method of
            the selected.
            :class:`providers <skosprovider.providers.VocabularyProvider>`.
        :param dict providers: Optional. If present, it should be a dictionary.
            This dictionary can contain any of the keyword arguments available
            to the :meth:`get_providers` method. The query will then only
            be passed to the providers confirming to these arguments.
        :param string language: Optional. If present, it should be a
            :term:`language-tag`. This language-tag is passed on to the
            underlying providers and used when selecting the label to display
            for each concept.
        :returns: a list of :class:`dict`.
            Each dict has two keys: id and concepts.
        '''
        if 'providers' not in kwargs:
            providers = self.get_providers()
        else:
            pargs = kwargs['providers']
            if isinstance(pargs, list):
                providers = self.get_providers(ids=pargs)
            else:
                providers = self.get_providers(**pargs)
        kwarguments = {}
        if 'language' in kwargs:
            kwarguments['language'] = kwargs['language']
        return [{'id': p.get_vocabulary_id(), 'concepts': p.find(query, **kwarguments)}
                for p in providers]

    def get_all(self, **kwargs):
        '''Get all concepts from all providers.

        .. code-block:: python

            # get all concepts in all providers.
            registry.get_all()

            # get all concepts in all providers.
            # If possible, display the results with a Dutch label.
            registry.get_all(language='nl')

        :param string language: Optional. If present, it should be a
            :term:`language-tag`. This language-tag is passed on to the
            underlying providers and used when selecting the label to display
            for each concept.

        :returns: a list of :class:`dict`.
            Each dict has two keys: id and concepts.
        '''
        kwarguments = {}
        if 'language' in kwargs:
            kwarguments['language'] = kwargs['language']
        return [{'id': p.get_vocabulary_id(), 'concepts': p.get_all(**kwarguments)}
                for p in self.providers.values()]

    def get_by_uri(self, uri):
        '''Get a concept or collection by its uri.

        Returns a single concept or collection if one exists with this uri.
        Returns False otherwise.

        :param string uri: The uri to find a concept or collection for.
        :rtype: :class:`skosprovider.skos.Concept` or
            :class:`skosprovider.skos.Collection`
        '''
        # Check if there's a provider that's more likely to have the URI
        csuris = [csuri for csuri in self.concept_scheme_uri_map.keys() if uri.startswith(csuri)]
        for csuri in csuris:
            c = self.get_provider(csuri).get_by_uri(uri)
            if c:
                return c
        # Check all providers
        for p in self.providers.values():
            c = p.get_by_uri(uri)
            if c:
                return c
        return False


1			# -- coding: utf-8 --
2
3			'''This module provides a registry for skos providers.
4
5			This registry helps us find providers during runtime. We can also apply some
6			operations to all or several providers at the same time.
7			'''
8
9			from __future__ import unicode_literals
10
11
12			class RegistryException(Exception):
13			pass
14
15
16			class Registry:
17			'''
18			This registry collects all skos providers.
19			'''
20
21			providers = {}
22			'''
23			Dictionary containing all providers, keyed by id.
24			'''
25
26			concept_scheme_uri_map = {}
27			'''
28			Dictionary mapping concept scheme uri's to vocabulary id's.
29			'''
30
31			def __init__(self):
32			self.providers = {}
33			self.concept_scheme_uri_map = {}
34
35			def register_provider(self, provider):
36			'''
37			Register a :class:`skosprovider.providers.VocabularyProvider`.
38
39			:param skosprovider.providers.VocabularyProvider provider: The provider
40			to register.
41			'''
42			if provider.get_vocabulary_id() in self.providers:
43			raise RegistryException(
44			'A provider with this id has already been registered.')
45			self.providers[provider.get_vocabulary_id()] = provider
46			self.concept_scheme_uri_map[provider.concept_scheme.uri] = provider.get_vocabulary_id()
47
48			def remove_provider(self, id):
49			'''
50			Remove the provider with the given id or :term:`URI`.
51
52			:param str id: The identifier for the provider.
53			:returns: A :class:`skosprovider.providers.VocabularyProvider` or
54			`False` if the id is unknown.
55			'''
56			if id in self.providers:
57			p = self.providers.get(id, False)
58			del self.providers[id]
59			del self.concept_scheme_uri_map[p.concept_scheme.uri]
60			return p
61			elif id in self.concept_scheme_uri_map:
62			id = self.concept_scheme_uri_map[id]
63			return self.remove_provider(id)
64			else:
65			return False
66
67			def get_provider(self, id):
68			'''
69			Get a provider by id or :term:`uri`.
70
71			:param str id: The identifier for the provider. This can either be the
72			id with which it was registered or the :term:`uri` of the conceptscheme
73			that the provider services.
74			:returns: A :class:`skosprovider.providers.VocabularyProvider`
75			or `False` if the id or uri is unknown.
76			'''
77			if id in self.providers:
78			return self.providers.get(id, False)
79			elif id in self.concept_scheme_uri_map:
80			return self.providers.get(self.concept_scheme_uri_map[id], False)
81			return False
82
83			def get_providers(self, **kwargs):
84			'''Get all providers registered.
85
86			If keyword `ids` is present, get only the providers with these ids.
87
88			If keys `subject` is present, get only the providers that have this subject.
89
90			.. code-block:: python
91
92			# Get all providers with subject 'biology'
93			registry.get_providers(subject='biology')
94
95			# Get all providers with id 1 or 2
96			registry.get_providers(ids=[1,2])
97
98			# Get all providers with id 1 or 2 and subject 'biology'
99			registry.get_providers(ids=[1,2], subject='biology']
100
101			:param list ids: Only return providers with one of the Ids or uris.
102			:param str subject: Only return providers with this subject.
103			:returns: A list of :class:`providers <skosprovider.providers.VocabularyProvider>`
104			'''
105			if 'ids' in kwargs:
106			ids = [self.concept_scheme_uri_map.get(id, id) for id in kwargs['ids']]
107			providers = [
108			self.providers[k] for k in self.providers.keys() if k in ids
109			]
110			else:
111			providers = list(self.providers.values())
112			if 'subject' in kwargs:
113			providers = [p for p in providers if kwargs['subject'] in p.metadata['subject']]
114			return providers
115
116			def find(self, query, **kwargs):
117			'''Launch a query across all or a selection of providers.
118
119			.. code-block:: python
120
121			# Find anything that has a label of church in any provider.
122			registry.find({'label': 'church'})
123
124			# Find anything that has a label of church with the BUILDINGS provider.
125			# Attention, this syntax was deprecated in version 0.3.0
126			registry.find({'label': 'church'}, providers=['BUILDINGS'])
127
128			# Find anything that has a label of church with the BUILDINGS provider.
129			registry.find({'label': 'church'}, providers={'ids': ['BUILDINGS']})
130
131			# Find anything that has a label of church with a provider
132			# marked with the subject 'architecture'.
133			registry.find({'label': 'church'}, providers={'subject': 'architecture'})
134
135			# Find anything that has a label of church in any provider.
136			# If possible, display the results with a Dutch label.
137			registry.find({'label': 'church'}, language='nl')
138
139			:param dict query: The query parameters that will be passed on to each
140			:meth:`~skosprovider.providers.VocabularyProvider.find` method of
141			the selected.
142			:class:`providers <skosprovider.providers.VocabularyProvider>`.
143			:param dict providers: Optional. If present, it should be a dictionary.
144			This dictionary can contain any of the keyword arguments available
145			to the :meth:`get_providers` method. The query will then only
146			be passed to the providers confirming to these arguments.
147			:param string language: Optional. If present, it should be a
148			:term:`language-tag`. This language-tag is passed on to the
149			underlying providers and used when selecting the label to display
150			for each concept.
151			:returns: a list of :class:`dict`.
152			Each dict has two keys: id and concepts.
153			'''
154			if 'providers' not in kwargs:
155			providers = self.get_providers()
156			else:
157			pargs = kwargs['providers']
158			if isinstance(pargs, list):
159			providers = self.get_providers(ids=pargs)
160			else:
161			providers = self.get_providers(**pargs)
162			kwarguments = {}
163			if 'language' in kwargs:
164			kwarguments['language'] = kwargs['language']
165			return [{'id': p.get_vocabulary_id(), 'concepts': p.find(query, **kwarguments)}
166			for p in providers]
167
168			def get_all(self, **kwargs):
169			'''Get all concepts from all providers.
170
171			.. code-block:: python
172
173			# get all concepts in all providers.
174			registry.get_all()
175
176			# get all concepts in all providers.
177			# If possible, display the results with a Dutch label.
178			registry.get_all(language='nl')
179
180			:param string language: Optional. If present, it should be a
181			:term:`language-tag`. This language-tag is passed on to the
182			underlying providers and used when selecting the label to display
183			for each concept.
184
185			:returns: a list of :class:`dict`.
186			Each dict has two keys: id and concepts.
187			'''
188			kwarguments = {}
189			if 'language' in kwargs:
190			kwarguments['language'] = kwargs['language']
191			return [{'id': p.get_vocabulary_id(), 'concepts': p.get_all(**kwarguments)}
192			for p in self.providers.values()]
193
194			def get_by_uri(self, uri):
195			'''Get a concept or collection by its uri.
196
197			Returns a single concept or collection if one exists with this uri.
198			Returns False otherwise.
199
200			:param string uri: The uri to find a concept or collection for.
201			:rtype: :class:`skosprovider.skos.Concept` or
202			:class:`skosprovider.skos.Collection`
203			'''
204			# Check if there's a provider that's more likely to have the URI
205			csuris = [csuri for csuri in self.concept_scheme_uri_map.keys() if uri.startswith(csuri)]
206			for csuri in csuris:
207			c = self.get_provider(csuri).get_by_uri(uri)
208			if c:
209			return c
210			# Check all providers
211			for p in self.providers.values():
212			c = p.get_by_uri(uri)
213			if c:
214			return c
215			return False
216

OnroerendErfgoed / skosprovider

Branch — master (d5006c)

Registry.get_provider() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like