Source code for keystone.common.provider_api

# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.


[docs] class ProviderAPIRegistry: __shared_object_state: dict = {} __registry: dict = {} __iter__ = __registry.__iter__ __getitem__ = __registry.__getitem__ locked = False def __init__(self): # NOTE(morgan): This rebinds __dict__ and allows all instances of # the provider API to share a common state. Any changes except # rebinding __dict__ will maintain the same state stored on the class # not the instance. This design pattern is preferable to # full singletons where state sharing is the important "feature" # derived from the "singleton" # # Use "super" to bypass the __setattr__ preventing changes to the # object itself. super().__setattr__('__dict__', self.__shared_object_state) def __getattr__(self, item): """Do attr lookup.""" try: return self.__registry[item] except KeyError: raise AttributeError(f"'ProviderAPIs' has no attribute {item}") def __setattr__(self, key, value): """Do not allow setting values on the registry object.""" raise RuntimeError( 'Programming Error: You may not set values on the ' 'ProviderAPIRegistry objects.' ) def _register_provider_api(self, name, obj): """Register an instance of a class as a provider api.""" if name == 'driver': raise ValueError('A provider may not be named "driver".') if self.locked: raise RuntimeError( 'Programming Error: The provider api registry has been ' 'locked (post configuration). Ensure all provider api ' 'managers are instantiated before locking.' ) if name in self.__registry: raise DuplicateProviderError( f'`{name}` has already been registered as an api ' f'provider by `{self.__registry[name]!r}`' ) self.__registry[name] = obj def _clear_registry_instances(self): """ONLY USED FOR TESTING.""" self.__registry.clear() # Use super to allow setting around class implementation of __setattr__ super().__setattr__('locked', False)
[docs] def lock_provider_registry(self): # Use super to allow setting around class implementation of __setattr__ super().__setattr__('locked', True)
[docs] def deferred_provider_lookup(self, api, method): """Create descriptor that performs lookup of api and method on demand. For specialized cases, such as the enforcer "get_member_from_driver" which needs to be effectively a "classmethod", this method returns a smart descriptor object that does the lookup at runtime instead of at import time. :param api: The api to use, e.g. "identity_api" :type api: str :param method: the method on the api to return :type method: str """ class DeferredProviderLookup: def __init__(self, api, method): self.__api = api self.__method = method def __get__(self, instance, owner): api = getattr(ProviderAPIs, self.__api) return getattr(api, self.__method) return DeferredProviderLookup(api, method)
[docs] class DuplicateProviderError(Exception): """Attempting to register a duplicate API provider."""
[docs] class ProviderAPIMixin: """Allow referencing provider apis on self via __getattr__. Be sure this class is first in the class definition for inheritance. """ def __getattr__(self, item): """Magic getattr method.""" try: return getattr(ProviderAPIs, item) except AttributeError: return self.__getattribute__(item)
ProviderAPIs = ProviderAPIRegistry()