Source code for skosprovider.registry
# -*- 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
[docs]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 = {}
[docs] 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()
[docs] 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
[docs] def get_provider(self, id):
'''
Get a provider by id.
: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:
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
[docs] 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']
'''
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
[docs] 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'})
: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 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>`.
:returns: a list of :class:`dict`.
Each dict has two keys: id and concepts.
'''
if not 'providers' 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)
return [{'id': p.get_vocabulary_id(), 'concepts': p.find(query)}
for p in providers]
[docs] def get_all(self):
'''Get all concepts from all providers.
:returns: a list of :class:`dict`.
Each dict has two keys: id and concepts.
'''
return [{'id': p.get_vocabulary_id(), 'concepts': p.get_all()}
for p in self.providers.values()]
[docs] 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)]
if len(csuris):
for csuri in csuris:
p = self.get_provider(csuri)
c = p.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