Skip to content

Commit

Permalink
Load all resource versions and sort them (#488)
Browse files Browse the repository at this point in the history
  • Loading branch information
jacobtomlinson authored Sep 10, 2024
1 parent c67c1c8 commit cd3236f
Show file tree
Hide file tree
Showing 3 changed files with 131 additions and 20 deletions.
29 changes: 16 additions & 13 deletions kr8s/_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
from cryptography import x509

from ._auth import KubeAuth
from ._data_utils import dict_to_selector
from ._data_utils import dict_to_selector, sort_versions
from ._exceptions import APITimeoutError, ServerError

if TYPE_CHECKING:
Expand Down Expand Up @@ -570,18 +570,21 @@ async def async_api_resources(self) -> list[dict]:
async with self.call_api(method="GET", version="", base="/apis") as response:
api_list = response.json()
for api in sorted(api_list["groups"], key=lambda d: d["name"]):
version = api["versions"][0]["groupVersion"]
async with self.call_api(
method="GET", version="", base="/apis", url=version
) as response:
resource = response.json()
resources.extend(
[
{"version": version, **r}
for r in resource["resources"]
if "/" not in r["name"]
]
)
for api_version in sort_versions(
api["versions"], key=lambda x: x["groupVersion"]
):
version = api_version["groupVersion"]
async with self.call_api(
method="GET", version="", base="/apis", url=version
) as response:
resource = response.json()
resources.extend(
[
{"version": version, **r}
for r in resource["resources"]
if "/" not in r["name"]
]
)
return resources

async def api_versions(self) -> AsyncGenerator[str]:
Expand Down
72 changes: 65 additions & 7 deletions kr8s/_data_utils.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,15 @@
# SPDX-FileCopyrightText: Copyright (c) 2023-2024, Kr8s Developers (See LICENSE for list)
# SPDX-License-Identifier: BSD 3-Clause License
"""Utilities for working with Kubernetes data structures."""
from typing import Any, Dict, List
from __future__ import annotations

import re
from typing import Any, Callable


def list_dict_unpack(
input_list: List[Dict], key: str = "key", value: str = "value"
) -> Dict:
input_list: list[dict], key: str = "key", value: str = "value"
) -> dict:
"""Convert a list of dictionaries to a single dictionary.
Args:
Expand All @@ -21,8 +24,8 @@ def list_dict_unpack(


def dict_list_pack(
input_dict: Dict, key: str = "key", value: str = "value"
) -> List[Dict]:
input_dict: dict, key: str = "key", value: str = "value"
) -> list[dict]:
"""Convert a dictionary to a list of dictionaries.
Args:
Expand All @@ -36,7 +39,7 @@ def dict_list_pack(
return [{key: k, value: v} for k, v in input_dict.items()]


def dot_to_nested_dict(dot_notated_key: str, value: Any) -> Dict:
def dot_to_nested_dict(dot_notated_key: str, value: Any) -> dict:
"""Convert a dot notated key to a nested dictionary.
Args:
Expand All @@ -57,7 +60,7 @@ def dot_to_nested_dict(dot_notated_key: str, value: Any) -> Dict:
return nested_dict


def dict_to_selector(selector_dict: Dict) -> str:
def dict_to_selector(selector_dict: dict) -> str:
"""Convert a dictionary to a Kubernetes selector.
Args:
Expand Down Expand Up @@ -102,3 +105,58 @@ def xdict(*in_dict, **kwargs):
if len(in_dict) == 1:
[kwargs] = in_dict
return {k: v for k, v in kwargs.items() if v is not None}


def sort_versions(
versions: list[Any], key: Callable = lambda x: x, reverse: bool = False
) -> list[Any]:
"""Sort a list of Kubernetes versions by priority.
Follows the spcification
https://kubernetes.io/docs/tasks/extend-kubernetes/custom-resources/custom-resource-definition-versioning/#version-priority
Args:
versions: A list of Kubernetes versions to sort.
key: A function to extract the version string from each element in the list.
Defaults to the identity function
reverse: If True, sort in descending order. Defaults to False
Returns:
A list of Kubernetes versions sorted by priority.
Examples:
>>> sort_versions(["v1", "v2", "v2beta1"])
["v2", "v1", "v2beta1"]
>>> sort_versions(["v1beta2", "foo1", "foo10", "v1"])
["v1", "v1beta2", "foo1", "foo10"]
"""
pattern = r"^v\d+((alpha|beta)\d+)?$"
stable = []
alphas = []
betas = []
others = []
for version in versions:
if re.match(pattern, key(version)) is not None:
if "alpha" in key(version):
alphas.append(version)
elif "beta" in key(version):
betas.append(version)
else:
stable.append(version)
else:
others.append(version)

stable = sorted(stable, key=lambda v: int(key(v)[1:]), reverse=True)
betas = sorted(
betas, key=lambda v: tuple(map(int, key(v)[1:].split("beta"))), reverse=True
)
alphas = sorted(
alphas, key=lambda v: tuple(map(int, key(v)[1:].split("alpha"))), reverse=True
)
others = sorted(others, key=lambda v: key(v))

output = stable + betas + alphas + others
if reverse:
output.reverse()
return output
50 changes: 50 additions & 0 deletions kr8s/tests/test_data_utils.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,15 @@
# SPDX-FileCopyrightText: Copyright (c) 2023-2024, Kr8s Developers (See LICENSE for list)
# SPDX-License-Identifier: BSD 3-Clause License
import random

import pytest

from kr8s._data_utils import (
dict_list_pack,
dict_to_selector,
dot_to_nested_dict,
list_dict_unpack,
sort_versions,
xdict,
)

Expand Down Expand Up @@ -70,3 +73,50 @@ def test_xdict():
"foo": "bar",
"baz": "qux",
}


def test_sort_version_priorities():
# Sample list based on list from Kubernetes documentation
# https://kubernetes.io/docs/tasks/extend-kubernetes/custom-resources/custom-resource-definition-versioning/#version-priority
versions = [
"v200",
"v10",
"v2",
"v1",
"v11beta2",
"v10beta3",
"v3beta1",
"v12alpha1",
"v11alpha2",
"v11alpha1",
"v3alpha1",
"12345",
"foo1",
"foo10",
"helloworld",
"vfoobaralpha1",
]

# Select some random permutations of the above list and ensure they get correctly sorted
sample = versions.copy() # Create a copy that we can shuffle
random.seed(42) # Ensure the same random order each time for deterministic tests
for _ in range(30):
random.shuffle(sample)
assert sort_versions(list(sample)) == versions


def test_sort_version_priorities_key():
versions = [
{"version": "v2"},
{"version": "v1"},
{"version": "v1beta2"},
{"version": "v1beta1"},
{"version": "v1alpha1"},
]

# Select some random permutations of the above list and ensure they get correctly sorted
sample = versions.copy() # Create a copy that we can shuffle
random.seed(42) # Ensure the same random order each time for deterministic tests
for _ in range(30):
random.shuffle(sample)
assert sort_versions(list(sample), key=lambda x: x["version"]) == versions

0 comments on commit cd3236f

Please sign in to comment.