Registry.get_by_uri() - Code Metrics - Inspection of "Merge pull request #28 from koenedaele/27_handle_N..." - OnroerendErfgoed/skosprovider - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 7c157d...b04ec2 )

by Koen

created 2016-07-27 16:49 UTC

Registry.get_by_uri() D

↳ Parent: Registry

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	8
c	1
b	0
f	0
dl	0
loc	25
rs	4

# -*- 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

from .uri import is_uri


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

OnroerendErfgoed / skosprovider

Push — master ( 7c157d...b04ec2 )

Registry.get_by_uri() D

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like