openstack
/
mistral-lib
Code Issues Proposed changes

Merge "Adds minimum common shared code for custom actions API"

This commit is contained in:
Jenkins 2017-03-15 09:53:14 +00:00 committed by Gerrit Code Review
parent a60de5c8f5 5764aec7ae
commit e8814640bb
17 changed files with 781 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
README.rst
View File

@ -11,11 +11,13 @@ Team and repository tags
mistral-lib
===========
Mistral shared routings and utilities (Actions API, YAQL functions API, data types etc.)
This library contains data types, exceptions, functions and utilities common to
Mistral, python-mistralclient and mistral-extra repositories. This library also
contains the public interfaces for 3rd party integration (e.g. Actions API, YAQL
functions API, etc.)
This Mistral subproject aims to have all common APIs, data types and utilities shared
across Mistral ecosystem and used to write Mistral pluggable extensions such as custom
actions and YAQL functions.
If you want to use OpenStack in your custom actions or functions, you will also
need to use http://git.openstack.org/cgit/openstack/mistral-extra .
* Free software: Apache license
* Documentation: http://docs.openstack.org/developer/mistral

47
doc/source/creating_custom_actions.rst Normal file
View File

@ -0,0 +1,47 @@
============================
How to write a Custom Action
============================
1. Write a class inherited from mistral.actions.Action
.. code-block:: python
from mistral_lib import actions
class RunnerAction(actions.Action):
def __init__(self, param):
# store the incoming params
self.param = param
def run(self):
# return your results here
return {'status': 0}
2. Publish the class in a namespace (in your ``setup.cfg``)
.. code-block:: ini
[entry_points]
mistral.actions =
example.runner = my.mistral_plugins.somefile:RunnerAction
3. Reinstall your library package if it was installed in system (not in virtualenv).
4. Run db-sync tool to ensure your actions are in Mistral's database
.. code-block:: console
$ mistral-db-manage --config-file <path-to-config> populate
5. Now you can call the action ``example.runner``
.. code-block:: yaml
my_workflow:
tasks:
my_action_task:
action: example.runner
input:
param: avalue_to_pass_in

17
mistral_lib/actions/__init__.py
View File

@ -0,0 +1,17 @@
# Copyright 2017 - Red Hat, Inc.
# All Rights Reserved.
#
# 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.
from mistral_lib.actions.base import Action
__all__ = ['Action']

47
mistral_lib/actions/api/execution.py
View File

@ -1,47 +0,0 @@
# Copyright 2016 - Nokia Networks.
#
# 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.
# TODO(rakhmerov): Add functions to get current execution info.
def get_workflow_name():
raise NotImplementedError
def get_workflow_execution_id():
raise NotImplementedError
def get_task_name():
raise NotImplementedError
def get_task_tags():
raise NotImplementedError
def get_task_execution_id():
raise NotImplementedError
def get_action_name():
raise NotImplementedError
def get_action_execution_id():
raise NotImplementedError
def get_mistral_callback_url():
raise NotImplementedError

27
mistral_lib/actions/api/security.py
View File

@ -1,27 +0,0 @@
# Copyright 2016 - Nokia Networks.
#
# 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.
# TODO(rakhmerov): Define necessary security related functions.
def get_user():
raise NotImplementedError
def get_project():
raise NotImplementedError
def get_auth_token():
raise NotImplementedError

36
mistral_lib/actions/api/types.py
View File

@ -1,36 +0,0 @@
# Copyright 2016 - Nokia Networks.
#
# 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.
# TODO(rakhmerov): Add necessary data types (and serializers, if needed).
class Result(object):
"""Explicit data structure containing a result of task execution."""
def __init__(self, data=None, error=None):
self.data = data
self.error = error
def __repr__(self):
return 'Result [data=%s, error=%s]' % (
repr(self.data), repr(self.error))
def is_error(self):
return self.error is not None
def is_success(self):
return not self.is_error()
def __eq__(self, other):
return self.data == other.data and self.error == other.error

29
mistral_lib/actions/api/utils.py
View File

@ -1,29 +0,0 @@
# Copyright 2016 - Nokia Networks.
#
# 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.
# TODO(rakhmerov): Add necessary utilities used by actions.
def get_action(name):
"""Gets an instance of action that can be executed on Mistral.
Given a name this method builds an instance of a certain action
that can be run in a regular way using Mistral execution mechanism
(routing to an executor etc.) by calling its run() method.
:param name: Action name.
"""
# TODO(rakhmerov): Implement.
raise NotImplementedError

24
mistral_lib/actions/api/base.py → mistral_lib/actions/base.py
View File

@ -1,4 +1,5 @@
# Copyright 2016 - Nokia Networks.
# Copyright 2017 - Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@ -30,17 +31,15 @@ class Action(object):
action classes may have any number of parameters defining action behavior.
These parameters must correspond to parameters declared in action
specification (e.g. using DSL or others).
Action initializer may have a conventional argument with name
"action_context". If it presents then action factory will fill it with
a dictionary containing contextual information like execution identifier,
workbook name and other that may be needed for some specific action
implementations.
"""
@abc.abstractmethod
def run(self):
def run(self, context):
"""Run action logic.
:param context: a dictionary containing contextual information like
execution identifier, workbook name and other data that may be needed
for some specific action implementations.
:return: Result of the action. Note that for asynchronous actions
it should always be None, however, if even it's not None it will be
ignored by a caller.
@ -59,19 +58,6 @@ class Action(object):
"""
pass
@abc.abstractmethod
def test(self):
"""Returns action test result.
This method runs in test mode as a test version of method run() to
generate and return a representative test result. It's basically a
contract for action 'dry-run' behavior specifically useful for
testing and workflow designing purposes.
:return: Representative action result.
"""
pass
def is_sync(self):
"""Returns True if the action is synchronous, otherwise False.

78
mistral_lib/actions/types.py Normal file
View File

@ -0,0 +1,78 @@
# Copyright 2014 - Mirantis, Inc.
# Copyright 2015 - StackStorm, Inc.
# Copyright 2016 - Brocade Communications Systems, Inc.
#
# 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.
from mistral_lib import serialization
from mistral_lib import utils
class Result(serialization.MistralSerializable):
"""Explicit data structure containing a result of task execution."""
def __init__(self, data=None, error=None, cancel=False):
self.data = data
self.error = error
self.cancel = cancel
def __repr__(self):
return 'Result [data=%s, error=%s, cancel=%s]' % (
repr(self.data), repr(self.error), str(self.cancel)
)
def cut_repr(self):
return 'Result [data=%s, error=%s, cancel=%s]' % (
utils.cut(self.data), utils.cut(self.error), str(self.cancel)
)
def is_cancel(self):
return self.cancel
def is_error(self):
return self.error is not None and not self.is_cancel()
def is_success(self):
return not self.is_error() and not self.is_cancel()
def __eq__(self, other):
return (
self.data == other.data and
self.error == other.error and
self.cancel == other.cancel
)
def __ne__(self, other):
return not self.__eq__(other)
def to_dict(self):
return ({'result': self.data}
if self.is_success() else {'result': self.error})
class ResultSerializer(serialization.DictBasedSerializer):
def serialize_to_dict(self, entity):
return {
'data': entity.data,
'error': entity.error,
'cancel': entity.cancel
}
def deserialize_from_dict(self, entity_dict):
return Result(
entity_dict['data'],
entity_dict['error'],
entity_dict.get('cancel', False)
)
serialization.register_serializer(Result, ResultSerializer())

82
mistral_lib/exceptions.py Normal file
View File

@ -0,0 +1,82 @@
# Copyright 2013 - Mirantis, Inc.
# Copyright 2015 - StackStorm, Inc.
# Copyright 2016 - Brocade Communications Systems, Inc.
# Copyright 2016 - Red Hat, Inc.
#
# 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.
class MistralExceptionBase(Exception):
"""Base class for Mistral specific errors and exceptions.
A common parent class to derive MistralError and MistralException
classes from.
"""
message = "An unknown error occurred"
http_code = 500
def __init__(self, message=None):
if message is not None:
self.message = message
super(MistralError, self).__init__(
'%d: %s' % (self.http_code, self.message))
@property
def code(self):
"""This is here for webob to read.
https://github.com/Pylons/webob/blob/master/webob/exc.py
"""
return self.http_code
def __str__(self):
return self.message
class MistralError(MistralExceptionBase):
"""Mistral specific error.
Reserved for situations that can't be automatically handled. When it occurs
it signals that there is a major environmental problem like invalid startup
configuration or implementation problem (e.g. some code doesn't take care
of certain corner cases). From architectural perspective it's pointless to
try to handle this type of problems except doing some finalization work
like transaction rollback, deleting temporary files etc.
"""
message = "An unknown error occurred"
class MistralException(Exception):
"""Mistral specific exception.
Reserved for situations that are not critical for program continuation.
It is possible to recover from this type of problems automatically and
continue program execution. Such problems may be related with invalid user
input (such as invalid syntax) or temporary environmental problems.
In case if an instance of a certain exception type bubbles up to API layer
then this type of exception it must be associated with an http code so it's
clear how to represent it for a client.
To correctly use this class, inherit from it and define a 'message' and
'http_code' properties.
"""
message = "An unknown exception occurred"
class ApplicationContextNotFoundException(MistralException):
http_code = 400
message = "Application context not found"

192
mistral_lib/serialization.py Normal file
View File

@ -0,0 +1,192 @@
# Copyright 2017 Nokia Networks.
#
# 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.
import abc
from oslo_serialization import jsonutils
_SERIALIZER = None
class Serializer(object):
"""Base interface for entity serializers.
A particular serializer knows how to convert a certain object
into a string and back from that string into an object whose
state is equivalent to the initial object.
"""
@abc.abstractmethod
def serialize(self, entity):
"""Converts the given object into a string.
:param entity: A object to be serialized.
:return String containing the state of the object in serialized form.
"""
raise NotImplementedError
@abc.abstractmethod
def deserialize(self, data_str):
"""Converts the given string into an object.
:param data_str: String containing the state of the object in
serialized form.
:return: An object.
"""
raise NotImplementedError
class DictBasedSerializer(Serializer):
"""Dictionary-based serializer.
It slightly simplifies implementing custom serializers by introducing
a contract based on dictionary. A serializer class extending this class
just needs to implement conversion from object into dict and from dict
to object. It doesn't need to convert into string and back as required
bye the base serializer contract. Conversion into string is implemented
once with regard to possible problems that may occur for collection and
primitive types as circular dependencies, correct date format etc.
"""
def serialize(self, entity):
if entity is None:
return None
entity_dict = self.serialize_to_dict(entity)
return jsonutils.dumps(
jsonutils.to_primitive(entity_dict, convert_instances=True)
)
def deserialize(self, data_str):
if data_str is None:
return None
entity_dict = jsonutils.loads(data_str)
return self.deserialize_from_dict(entity_dict)
@abc.abstractmethod
def serialize_to_dict(self, entity):
raise NotImplementedError
@abc.abstractmethod
def deserialize_from_dict(self, entity_dict):
raise NotImplementedError
class MistralSerializable(object):
"""A mixin to generate a serialization key for a custom object."""
@classmethod
def get_serialization_key(cls):
return "%s.%s" % (cls.__module__, cls.__name__)
class PolymorphicSerializer(Serializer):
"""Polymorphic serializer.
The purpose of this class is to server as a serialization router
between serializers that can work with entities of particular type.
All concrete serializers associated with concrete entity classes
should be registered via method 'register', after that an instance
of polymorphic serializer can be used as a universal serializer
for an RPC system or something else.
When converting an object into a string this serializer also writes
a special key into the result string sequence so that it's possible
to find a proper serializer when deserializing this object.
If a primitive value is given as an entity this serializer doesn't
do anything special and simply converts a value into a string using
jsonutils. Similar when it converts a string into a primitive value.
"""
def __init__(self):
# {serialization key: serializer}
self.serializers = {}
@staticmethod
def _get_serialization_key(entity_cls):
if issubclass(entity_cls, MistralSerializable):
return entity_cls.get_serialization_key()
return None
def register(self, entity_cls, serializer):
key = self._get_serialization_key(entity_cls)
if not key:
return
if key in self.serializers:
raise RuntimeError(
"A serializer for the entity class has already been"
" registered: %s" % entity_cls
)
self.serializers[key] = serializer
def cleanup(self):
self.serializers.clear()
def serialize(self, entity):
if entity is None:
return None
key = self._get_serialization_key(type(entity))
# Primitive or not registered type.
if not key:
return jsonutils.dumps(
jsonutils.to_primitive(entity, convert_instances=True)
)
serializer = self.serializers.get(key)
result = {
'__serial_key': key,
'__serial_data': serializer.serialize(entity)
}
return jsonutils.dumps(result)
def deserialize(self, data_str):
if data_str is None:
return None
data = jsonutils.loads(data_str)
if isinstance(data, dict) and '__serial_key' in data:
serializer = self.serializers.get(data['__serial_key'])
return serializer.deserialize(data['__serial_data'])
return data
def get_polymorphic_serializer():
global _SERIALIZER
if _SERIALIZER is None:
_SERIALIZER = PolymorphicSerializer()
return _SERIALIZER
def register_serializer(entity_cls, serializer):
get_polymorphic_serializer().register(entity_cls, serializer)
def cleanup():
get_polymorphic_serializer().cleanup()

0
mistral_lib/actions/api/__init__.py → mistral_lib/tests/actions/__init__.py
View File

28
mistral_lib/tests/test_mistral_actions_api.py → mistral_lib/tests/actions/test_base.py
View File

@ -1,23 +1,31 @@
# Copyright 2017 Red Hat, Inc.
# All Rights Reserved.
#
# 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
# 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.
"""
Stub module for Custom Actions API tests.
"""
from mistral_lib.tests import base
from mistral_lib import actions
from mistral_lib.tests import base as tests_base
class TestActionsAPI(base.TestCase):
class TestAction(actions.Action):
def test_something(self):
pass
def run(self, context):
return context
class TestActionsBase(tests_base.TestCase):
def test_run(self):
context = {"name": "test"}
action = TestAction()
result = action.run(context)
assert result == context

108
mistral_lib/tests/test_serialization.py Normal file
View File

@ -0,0 +1,108 @@
# Copyright 2017 - Nokia Networks.
#
# 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.
from mistral_lib import serialization
from mistral_lib.tests import base
class MyClass(serialization.MistralSerializable):
def __init__(self, a, b):
self.a = a
self.b = b
def __eq__(self, other):
if not isinstance(other, MyClass):
return False
return other.a == self.a and other.b == self.b
class MyClassSerializer(serialization.DictBasedSerializer):
def serialize_to_dict(self, entity):
return {'a': entity.a, 'b': entity.b}
def deserialize_from_dict(self, entity_dict):
return MyClass(entity_dict['a'], entity_dict['b'])
class SerializationTest(base.TestCase):
def setUp(self):
super(SerializationTest, self).setUp()
serialization.register_serializer(MyClass, MyClassSerializer())
self.addCleanup(serialization.cleanup)
def test_dict_based_serializer(self):
obj = MyClass('a', 'b')
serializer = MyClassSerializer()
s = serializer.serialize(obj)
self.assertEqual(obj, serializer.deserialize(s))
self.assertIsNone(serializer.serialize(None))
self.assertIsNone(serializer.deserialize(None))
def test_polymorphic_serializer_primitive_types(self):
serializer = serialization.get_polymorphic_serializer()
self.assertEqual(17, serializer.deserialize(serializer.serialize(17)))
self.assertEqual(
0.34,
serializer.deserialize(serializer.serialize(0.34))
)
self.assertEqual(-5, serializer.deserialize(serializer.serialize(-5)))
self.assertEqual(
-6.3,
serializer.deserialize(serializer.serialize(-6.3))
)
self.assertFalse(serializer.deserialize(serializer.serialize(False)))
self.assertTrue(serializer.deserialize(serializer.serialize(True)))
self.assertEqual(
'abc',
serializer.deserialize(serializer.serialize('abc'))
)
self.assertEqual(
{'a': 'b', 'c': 'd'},
serializer.deserialize(serializer.serialize({'a': 'b', 'c': 'd'}))
)
self.assertEqual(
['a', 'b', 'c'],
serializer.deserialize(serializer.serialize(['a', 'b', 'c']))
)
def test_polymorphic_serializer_custom_object(self):
serializer = serialization.get_polymorphic_serializer()
obj = MyClass('a', 'b')
s = serializer.serialize(obj)
self.assertIn('__serial_key', s)
self.assertIn('__serial_data', s)
self.assertEqual(obj, serializer.deserialize(s))
self.assertIsNone(serializer.serialize(None))
self.assertIsNone(serializer.deserialize(None))
def test_register_twice(self):
self.assertRaises(
RuntimeError,
serialization.register_serializer,
MyClass,
MyClassSerializer()
)

86
mistral_lib/tests/test_utils.py Normal file
View File

@ -0,0 +1,86 @@
# Copyright 2013 - Mirantis, Inc.
# Copyright 2017 - Nokia Networks.
# Copyright 2017 - Red Hat, Inc.
# All Rights Reserved.
#
# 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.
from mistral_lib.tests import base as tests_base
from mistral_lib import utils
class TestUtils(tests_base.TestCase):
def test_cut_string(self):
s = 'Hello, Mistral!'
self.assertEqual('Hello...', utils.cut_string(s, length=5))
self.assertEqual(s, utils.cut_string(s, length=100))
def test_cut_list(self):
l = ['Hello, Mistral!', 'Hello, OpenStack!']
self.assertEqual("['Hello, M...", utils.cut_list(l, 11))
self.assertEqual("['Hello, Mistr...", utils.cut_list(l, 15))
self.assertEqual("['Hello, Mistral!', 'He...", utils.cut_list(l, 24))
self.assertEqual(
"['Hello, Mistral!', 'Hello, OpenStack!']",
utils.cut_list(l, 100)
)
self.assertEqual("[1, 2...", utils.cut_list([1, 2, 3, 4, 5], 4))
self.assertEqual("[1, 2...", utils.cut_list([1, 2, 3, 4, 5], 5))
self.assertEqual("[1, 2, 3...", utils.cut_list([1, 2, 3, 4, 5], 6))
self.assertRaises(ValueError, utils.cut_list, (1, 2))
def test_cut_dict_with_strings(self):
d = {'key1': 'value1', 'key2': 'value2'}
s = utils.cut_dict(d, 9)
self.assertIn(s, ["{'key1': '...", "{'key2': '..."])
s = utils.cut_dict(d, 13)
self.assertIn(s, ["{'key1': 'va...", "{'key2': 'va..."])
s = utils.cut_dict(d, 19)
self.assertIn(
s,
["{'key1': 'value1', ...", "{'key2': 'value2', ..."]
)
self.assertIn(
utils.cut_dict(d, 100),
[
"{'key1': 'value1', 'key2': 'value2'}",
"{'key2': 'value2', 'key1': 'value1'}"
]
)
def test_cut_dict_with_digits(self):
d = {1: 2, 3: 4}
s = utils.cut_dict(d, 6)
self.assertIn(s, ["{1: 2, ...", "{3: 4, ..."])
s = utils.cut_dict(d, 8)
self.assertIn(s, ["{1: 2, 3...", "{3: 4, 1..."])
s = utils.cut_dict(d, 100)
self.assertIn(s, ["{1: 2, 3: 4}", "{3: 4, 1: 2}"])

141
mistral_lib/utils/__init__.py Normal file
View File

@ -0,0 +1,141 @@
# Copyright 2013 - Mirantis, Inc.
# Copyright 2017 - Nokia Networks.
# Copyright 2017 - Red Hat, Inc.
# All Rights Reserved.
#
# 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.
def cut_dict(d, length=100):
"""Truncates string representation of a dictionary for a given length.
:param d: dictionary to truncate
:param length: amount of characters to truncate to
:return: string containing given length of characters from the dictionary
"""
if not isinstance(d, dict):
raise ValueError("A dictionary is expected, got: %s" % type(d))
res = "{"
idx = 0
for key, value in d.items():
k = str(key)
v = str(value)
# Processing key.
new_len = len(res) + len(k)
is_str = isinstance(key, str)
if is_str:
new_len += 2
if new_len >= length:
res += "'%s..." % k[:length - new_len] if is_str else "%s..." % k
break
else:
res += "'%s'" % k if is_str else k
res += ": "
# Processing value.
new_len = len(res) + len(v)
is_str = isinstance(value, str)
if is_str:
new_len += 2
if new_len >= length:
res += "'%s..." % v[:length - new_len] if is_str else "%s..." % v
break
else:
res += "'%s'" % v if is_str else v
res += ', ' if idx < len(d) - 1 else '}'
if len(res) >= length:
res += '...'
break
idx += 1
return res
def cut_list(l, length=100):
"""Truncates string representation of a list for a given length.
:param l: list to truncate
:param length: amount of characters to truncate to
:return: string containing given length of characters from the list
"""
if not isinstance(l, list):
raise ValueError("A list is expected, got: %s" % type(l))
res = '['
for idx, item in enumerate(l):
s = str(item)
new_len = len(res) + len(s)
is_str = isinstance(item, str)
if is_str:
new_len += 2
if new_len >= length:
res += "'%s..." % s[:length - new_len] if is_str else "%s..." % s
break
else:
res += "'%s'" % s if is_str else s
res += ', ' if idx < len(l) - 1 else ']'
return res
def cut_string(s, length=100):
"""Truncates a string for a given length.
:param s: string to truncate
:param length: amount of characters to truncate to
:return: string containing given length of characters
"""
if len(s) > length:
return "%s..." % s[:length]
return s
def cut(data, length=100):
"""Truncates string representation of data for a given length.
:param data: a dictionary, list or string to truncate
:param length: amount of characters to truncate to
:return: string containing given length of characters
"""
if not data:
return data
if isinstance(data, list):
return cut_list(data, length=length)
if isinstance(data, dict):
return cut_dict(data, length=length)
return cut_string(str(data), length=length)

1
requirements.txt
View File

@ -4,3 +4,4 @@
pbr>=2.0.0 # Apache-2.0
Babel>=2.3.4 # BSD
oslo.serialization>=1.10.0 # Apache-2.0