4fc1b9c424
six.reraise can be converted into "raise new_exc from original_exc" introduced in python3. In case of horizon.exceptions.handle(), six.reraise was used to raise the original exception again, so we can convert it into "raise". six is removed from requirements.txt. Note that we need to keep six in lower-constraints.txt as dependent libraries still use six. Change-Id: I0672a516083727c245f1d4fa5b10eed9cd81a726
212 lines
7.2 KiB
Python
212 lines
7.2 KiB
Python
# 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.
|
|
|
|
"""
|
|
A set of custom types for oslo.config.
|
|
"""
|
|
|
|
import ast
|
|
import os
|
|
import re
|
|
|
|
from django.utils import encoding
|
|
from django.utils import functional
|
|
from django.utils.module_loading import import_string
|
|
from django.utils.translation import pgettext_lazy
|
|
from oslo_config import types
|
|
|
|
|
|
class Maybe(types.ConfigType):
|
|
"""A custom option type for a value that may be None."""
|
|
|
|
def __init__(self, type_):
|
|
self.type_ = type_
|
|
type_name = getattr(type_, 'type_name', 'unknown value')
|
|
super(Maybe, self).__init__('optional %s' % type_name)
|
|
|
|
def __call__(self, value):
|
|
if value is None:
|
|
return None
|
|
return self.type_(value)
|
|
|
|
def _formatter(self, value):
|
|
if value is None:
|
|
return ''
|
|
return self.type_._formatter(value)
|
|
|
|
|
|
class URL(types.ConfigType):
|
|
"""A custom option type for a URL or part of URL."""
|
|
|
|
CLEAN_SLASH_RE = re.compile(r'(?<!:)//')
|
|
|
|
def __init__(self):
|
|
super(URL, self).__init__('web URL')
|
|
|
|
def __call__(self, value):
|
|
if not isinstance(value, str):
|
|
raise ValueError("Expected URL.")
|
|
value = re.sub(self.CLEAN_SLASH_RE, '/', value)
|
|
if not value.endswith('/'):
|
|
value += '/'
|
|
return value
|
|
|
|
def _formatter(self, value):
|
|
return self.quote_trailing_and_leading_space(value)
|
|
|
|
|
|
class Path(types.ConfigType):
|
|
"""A custom option type for a path to file."""
|
|
|
|
def __init__(self):
|
|
super(Path, self).__init__('filesystem path')
|
|
|
|
def __call__(self, value):
|
|
if not isinstance(value, str):
|
|
raise ValueError("Expected file path.")
|
|
return os.path.normpath(value)
|
|
|
|
def _formatter(self, value):
|
|
return self.quote_trailing_and_leading_space(value)
|
|
|
|
|
|
class Translate(types.ConfigType):
|
|
"""A custom option type for translatable strings."""
|
|
|
|
def __init__(self, hint=None):
|
|
self.hint = hint
|
|
super(Translate, self).__init__('translatable string')
|
|
|
|
def __call__(self, value):
|
|
if not isinstance(value, str):
|
|
return value
|
|
return pgettext_lazy(value, self.hint)
|
|
|
|
def _formatter(self, value):
|
|
return self.quote_trailing_and_leading_space(
|
|
encoding.force_text(value))
|
|
|
|
|
|
class Literal(types.ConfigType):
|
|
"""A custom option type for a Python literal."""
|
|
|
|
def __init__(self, spec=None):
|
|
self.spec = spec
|
|
super(Literal, self).__init__('python literal')
|
|
|
|
def __call__(self, value):
|
|
if isinstance(value, str):
|
|
try:
|
|
value = ast.literal_eval(value)
|
|
except SyntaxError as e:
|
|
raise ValueError from e
|
|
self.validate(value, self.spec)
|
|
return self.update(value, self.spec)
|
|
|
|
def validate(self, result, spec): # noqa Yes, it's too complex.
|
|
"""Validate that the result has the correct structure."""
|
|
if spec is None:
|
|
# None matches anything.
|
|
return
|
|
if isinstance(spec, dict):
|
|
if not isinstance(result, dict):
|
|
raise ValueError('Dictionary expected, but %r found.' % result)
|
|
if spec:
|
|
spec_value = next(iter(spec.values())) # Yay Python 3!
|
|
for value in result.values():
|
|
self.validate(value, spec_value)
|
|
spec_key = next(iter(spec.keys()))
|
|
for key in result:
|
|
self.validate(key, spec_key)
|
|
if isinstance(spec, list):
|
|
if not isinstance(result, list):
|
|
raise ValueError('List expected, but %r found.' % result)
|
|
if spec:
|
|
for value in result:
|
|
self.validate(value, spec[0])
|
|
if isinstance(spec, tuple):
|
|
if not isinstance(result, tuple):
|
|
raise ValueError('Tuple expected, but %r found.' % result)
|
|
if len(result) != len(spec):
|
|
raise ValueError('Expected %d elements in tuple %r.' %
|
|
(len(spec), result))
|
|
for s, value in zip(spec, result):
|
|
self.validate(value, s)
|
|
if isinstance(spec, str):
|
|
if not isinstance(result, str):
|
|
raise ValueError('String expected, but %r found.' % result)
|
|
if isinstance(spec, int):
|
|
if not isinstance(result, int):
|
|
raise ValueError('Integer expected, but %r found.' % result)
|
|
if isinstance(spec, bool):
|
|
if not isinstance(result, bool):
|
|
raise ValueError('Boolean expected, but %r found.' % result)
|
|
|
|
def update(self, result, spec):
|
|
"""Replace elements with results of calling callables."""
|
|
if isinstance(spec, dict):
|
|
if spec:
|
|
spec_value = next(iter(spec.values()))
|
|
for key, value in result.items():
|
|
result[key] = self.update(value, spec_value)
|
|
if isinstance(spec, list):
|
|
if spec:
|
|
for i, value in enumerate(result):
|
|
result[i] = self.update(value, spec[0])
|
|
if isinstance(spec, tuple):
|
|
return tuple(self.update(value, s)
|
|
for s, value in zip(spec, result))
|
|
if callable(spec):
|
|
return spec(result)
|
|
return result
|
|
|
|
def _format(self, result):
|
|
if isinstance(result, dict):
|
|
return '{%s}' % ', '.join(
|
|
'%s: %s' % (key, self._format(value))
|
|
for key, value in result.items()
|
|
)
|
|
if isinstance(result, list):
|
|
return '[%s]' % ', '.join(self._format(value) for value in result)
|
|
if isinstance(result, tuple):
|
|
return '(%s)' % ', '.join(self._format(value) for value in result)
|
|
if isinstance(result, functional.Promise):
|
|
# Lazy translatable string.
|
|
return repr(encoding.force_text(result))
|
|
return repr(result)
|
|
|
|
def _formatter(self, value):
|
|
# We need to walk the lists and dicts to handle the Django lazy
|
|
# translatable strings inside.
|
|
return self._format(value)
|
|
|
|
|
|
class Importable(types.ConfigType):
|
|
"""A custom option type for an importable python object."""
|
|
|
|
def __init__(self):
|
|
super(Importable, self).__init__('importable python object')
|
|
|
|
def __call__(self, value):
|
|
if not isinstance(value, str):
|
|
# Already imported.
|
|
return value
|
|
try:
|
|
return import_string(value)
|
|
except ImportError as e:
|
|
raise ValueError from e
|
|
|
|
def _formatter(self, value):
|
|
module = value.__module__
|
|
name = value.__name__
|
|
return self.quote_trailing_and_leading_space('%s.%s' % (module, name))
|