servala/servala-portal
3
0
Fork 0
Code Issues 21 Pull requests 1 Projects 1 Releases Packages 2 Activity Actions

Simplify CRD and schema retrieval

Browse source
This commit is contained in:
Tobias Kunze 2025-03-26 15:26:14 +01:00
parent eaa0614839
commit ebf88527fe
1 changed files with 15 additions and 70 deletions
Download patch file Download diff file Expand all files Collapse all files

85
src/servala/core/models/service.py
View file

@ -318,81 +318,26 @@ class ServiceOfferingControlPlane(models.Model):
version = self.service_definition.api_definition["version"]
client = self.control_plane.get_kubernetes_client()
# We have to determine the ``name`` first, via the discovery API
# The official kubernetes client APIs only include the regular discovery API,
# but not the APIGroupDiscoveryList that we actually need, and passing it as
# response_type leads to the client receiving the data, but refusing to return
# it since the model is not defined. Instead, we have to manually
# construct the API call:
content_type = "application/json"
response_type = "APIGroupDiscoveryList"
accept_header = ",".join(
[
f"{content_type};g=apidiscovery.k8s.io;v={api_version};as={response_type}"
for api_version in ("v2", "v2beta1")
]
)
accept_header = f"{accept_header},{content_type}"
groups = client.call_api(
"/apis",
"GET",
header_params={"Accept": accept_header},
auth_settings=["BearerToken"],
_return_http_data_only=True,
_preload_content=False,
).json()
api_group = [
g for g in groups.get("items", []) if g["metadata"]["name"] == group
][0]
api_version = [v for v in api_group["versions"] if v["version"] == version][0]
resource = [
r for r in api_version["resources"] if r["responseKind"]["kind"] == kind
][0]
# This is the thing we went to all that effort for!
name = f"{resource['resource']}.{group}"
extensions_api = kubernetes.client.ApiextensionsV1Api(client)
crd = extensions_api.read_custom_resource_definition(name)
return crd
crds = extensions_api.list_custom_resource_definition()
matching_crd = None
for crd in crds.items:
if matching_crd:
break
if crd.spec.group == group:
for served_version in crd.spec.versions:
if served_version.name == version and served_version.served:
if crd.spec.names.kind == kind:
matching_crd = crd
break
return matching_crd
@cached_property
def resource_schema(self):
# We extract the schema directly from the API. As an alternative, you can
# also get it using self.resource_definition, though that requires some
# extra serialization to remove the API client types:
# for version in self.resource_definition.spec.versions:
# if self.service_definition.api_definition["version"] != version.name:
# continue
# if version.schema and version.schema.open_apiv3_schema:
# schema_dict = kubernetes.client.ApiClient().sanitize_for_serialization(
# version.schema.open_apiv3_schema
# )
# return schema_dict
kind = self.service_definition.api_definition["kind"]
group = self.service_definition.api_definition["group"]
version = self.service_definition.api_definition["version"]
client = self.control_plane.get_kubernetes_client()
response = client.call_api(
f"/openapi/v3/apis/{group}/{version}",
"GET",
header_params={"Accept": "application/json"},
auth_settings=["BearerToken"],
_return_http_data_only=True,
_preload_content=False,
).json()
for schema in response["components"]["schemas"].values():
gkvs = schema.get("x-kubernetes-group-version-kind")
if not gkvs or not isinstance(gkvs, list):
continue
if any(
[
gkv["group"] == group
and gkv["version"] == version
and gkv["kind"] == kind
for gkv in gkvs
]
):
return schema
for v in self.resource_definition.spec.versions:
if v.name == version:
return v.schema.open_apiv3_schema.to_dict()
@cached_property
def django_model(self):