Skip to content

Utils

Quantity

Convert quantity string to decimal

K8s converts user input quantities to "canonical form":

Before serializing, Quantity will be put in "canonical form". This means that Exponent/suffix will be adjusted up or down (with a corresponding increase or decrease in Mantissa) such that: a. No precision is lost b. No fractional digits will be emitted c. The exponent (or suffix) is as large as possible. The sign will be omitted unless the number is negative.

Examples: 1.5 will be serialized as "1500m" 1.5Gi will be serialized as "1536Mi"

Additional examples:

User input K8s representation
{"memory": "0.9Gi"} {"memory": "966367641600m"}
{"cpu": "0.30000000000000004"} {"cpu": "301m"}

You may need to compare different quantities when interacting with K8s.

Interface

lightkube.utils.quantity.parse_quantity(quantity: Optional[str]) -> Optional[decimal.Decimal]

Parse a quantity string into a bare (suffix-less) decimal.

Kubernetes converts user input to a canonical representation. For example, "0.9Gi" would be converted to "966367641600m". This function can be useful for comparing user input to actual values, for example comparing resource limits between a StatefulSet's template (statefulset.spec.template.spec.containers[i].resources) and a scheduled pod (pod.spec.containers[i].resources) after patching the StatefulSet.

Parameters

  • quantity str - An str representing a K8s quantity (e.g. "1Gi" or "1G"), per https://kubernetes.io/docs/reference/kubernetes-api/common-definitions/quantity/.

returns An instance of decimal.Decimal representing the quantity as a bare decimal.

Source code in src/lightkube/utils/quantity.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def parse_quantity(quantity: Optional[str]) -> Optional[decimal.Decimal]:
    """Parse a quantity string into a bare (suffix-less) decimal.

    Kubernetes converts user input to a canonical representation. For example, "0.9Gi" would be converted
    to "966367641600m".
    This function can be useful for comparing user input to actual values, for example comparing
    resource limits between a StatefulSet's template
    (`statefulset.spec.template.spec.containers[i].resources`) and a scheduled pod
    (`pod.spec.containers[i].resources`) after patching the StatefulSet.

    **Parameters**

    * **quantity** `str` - An str representing a K8s quantity (e.g. "1Gi" or "1G"), per
    https://kubernetes.io/docs/reference/kubernetes-api/common-definitions/quantity/.

    **returns**  An instance of `decimal.Decimal` representing the quantity as a bare decimal.
    """
    if quantity is None:
        # This is useful for comparing e.g. ResourceRequirements.limits.get("cpu"), which can be
        # None.
        return None

    pat = re.compile(r"([+-]?\d+(?:[.]\d*)?(?:e[+-]?\d+)?|[.]\d+(?:e[+-]?\d+)?)(.*)")
    match = pat.match(quantity)

    if not match:
        raise ValueError("Invalid quantity string: '{}'".format(quantity))

    try:
        value = decimal.Decimal(match.group(1))
    except ArithmeticError as e:
        raise ValueError("Invalid numerical value") from e

    unit = match.group(2)

    try:
        multiplier = MULTIPLIERS[unit]
    except KeyError as err:
        raise ValueError("Invalid unit suffix: {}".format(unit)) from err

    try:
        as_decimal = value * multiplier
        return as_decimal.quantize(decimal.Decimal("0.001"), rounding=decimal.ROUND_UP)
    except ArithmeticError as e:
        raise ValueError("Invalid numerical value") from e

lightkube.utils.quantity.equals_canonically(first, second) 

equals_canonically(
    first: ResourceRequirements,
    second: ResourceRequirements,
) -> bool

equals_canonically(
    first: Optional[dict], second: Optional[dict]
) -> bool

Compare two resource requirements for numerical equality.

Both arguments must be of the same type and can be either:

  • An instance of core_v1.ResourceRequirements
  • Optional[dict]: representing the "limits" or the "requests" portion of ResourceRequirements.
>>> equals_canonically({"cpu": "0.6"}, {"cpu": "600m"})
True

>>> equals_canonically(
        ResourceRequirements(limits={"cpu": "0.6"}),
        ResourceRequirements(limits={"cpu": "600m"})
    )
True

Parameters

  • first ResourceRequirements or dict - The first item to compare.
  • second ResourceRequirements or dict - The second item to compare.

returns True, if both arguments are numerically equal; False otherwise.

Source code in src/lightkube/utils/quantity.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def equals_canonically(first, second):
    """Compare two resource requirements for numerical equality.

    Both arguments must be of the same type and can be either:

    - An instance of `core_v1.ResourceRequirements`
    - `Optional[dict]`: representing the "limits" or the "requests" portion of `ResourceRequirements`.

    ```python
    >>> equals_canonically({"cpu": "0.6"}, {"cpu": "600m"})
    True

    >>> equals_canonically(
            ResourceRequirements(limits={"cpu": "0.6"}),
            ResourceRequirements(limits={"cpu": "600m"})
        )
    True
    ```

    **Parameters**

    * **first** `ResourceRequirements` or `dict` - The first item to compare.
    * **second** `ResourceRequirements` or `dict` - The second item to compare.

    **returns**  True, if both arguments are numerically equal; False otherwise.
    """
    if isinstance(first, (dict, type(None))) and isinstance(second, (dict, type(None))):
        # Args are 'limits' or 'requests' dicts
        return _equals_canonically(first, second)
    elif isinstance(first, ResourceRequirements) and isinstance(second, ResourceRequirements):
        # Args are ResourceRequirements, which may contain 'limits' and 'requests' dicts
        ks = ("limits", "requests")
        return all(_equals_canonically(getattr(first, k), getattr(second, k)) for k in ks)
    else:
        raise TypeError(
            "unsupported operand type(s) for canonical comparison: '{}' and '{}'".format(
                first.__class__.__name__,
                second.__class__.__name__,
            )
        )

Examples

Compare container memory request with limit

from lightkube.utils.quantity import parse_quantity

pod = client.get(Pod, name='my-pod')
container_res = pod.spec.containers[0].resources
if parse_quantity(container_res.requests['memory']) < parse_quantity(container_res.limits['memory']):
    ... # request is less than limit, do something ...

Compare container request with limit

from lightkube.utils.quantity import equals_canonically

pod = client.get(Pod, name='my-pod')
container_res = pod.spec.containers[0].resources
if equals_canonically(container_res.requests, container_res.limits):
    ... # requests and limits are the same ...

Complete example

After patching a statefulset's resource limits you may want to compare user's input to the statefulset's template to the active podspec:

>>> from lightkube import Client
>>> from lightkube.models.apps_v1 import StatefulSetSpec
>>> from lightkube.models.core_v1 import (Container, PodSpec, PodTemplateSpec, ResourceRequirements)
>>> from lightkube.resources.apps_v1 import StatefulSet
>>> from lightkube.resources.core_v1 import Pod
>>> from lightkube.types import PatchType
>>>
>>> resource_reqs = ResourceRequirements(
...     limits={"cpu": "0.8", "memory": "0.9Gi"},
...     requests={"cpu": "0.4", "memory": "0.5Gi"},
... )
>>>
>>> client = Client()
>>> statefulset = client.get(StatefulSet, name="prom")
>>>
>>> delta = StatefulSet(
...     spec=StatefulSetSpec(
...         selector=statefulset.spec.selector,
...         serviceName=statefulset.spec.serviceName,
...         template=PodTemplateSpec(
...             spec=PodSpec(
...                 containers=[Container(name="prometheus", resources=resource_reqs)]
...             )
...         )
...     )
... )
>>>
>>> client.patch(StatefulSet, "prom", delta, patch_type=PatchType.APPLY, field_manager="just me")
>>> client.get(StatefulSet, name="prom").spec.template.spec.containers[1].resources
ResourceRequirements(limits={'cpu': '800m', 'memory': '966367641600m'}, requests={'cpu': '400m', 'memory': '512Mi'})
>>>
>>> pod = client.get(Pod, name="prom-0")
>>> pod.spec.containers[1].resources
ResourceRequirements(limits={'cpu': '800m', 'memory': '966367641600m'}, requests={'cpu': '400m', 'memory': '512Mi'})
>>>
>>> from lightkube.utils.quantity import parse_quantity
>>> parse_quantity(pod.spec.containers[1].resources.requests["memory"])
Decimal('536870912.000')
>>> parse_quantity(pod.spec.containers[1].resources.requests["memory"]) == parse_quantity(resource_reqs.requests["memory"])
True
>>>
>>> from lightkube.utils.quantity import equals_canonically
>>> equals_canonically(pod.spec.containers[1].resources.limits, resource_reqs.limits)
True
>>> equals_canonically(pod.spec.containers[1].resources, resource_reqs)
True