pipeline fixes

icekube/models/base.py

@@ -10,7 +10,7 @@
 from icekube.relationships import Relationship
 from icekube.utils import to_camel_case
 from kubernetes import client
-from pydantic import BaseModel, Field, model_validator, computed_field
+from pydantic import BaseModel, Field, computed_field, model_validator
 
 logger = logging.getLogger(__name__)
 
@@ -56,9 +56,10 @@ def __eq__(self, other) -> bool:
  def data(self) -> Dict[str, Any]:
  return cast(Dict[str, Any], json.loads(self.raw or "{}"))
 
- @computed_field
+ @computed_field # type: ignore
+ @property
  def labels(self) -> Dict[str, str]:
- return self.data.get("metadata", {}).get("labels", {})
+ return cast(Dict[str, str], self.data.get("metadata", {}).get("labels", {}))
 
  @model_validator(mode="before")
  def inject_missing_required_fields(cls, values):

icekube/models/cluster.py

@@ -1,6 +1,5 @@
 import re
-
-from typing import Dict, List
+from typing import Any, Dict, List
 
 from icekube.models.base import RELATIONSHIP, Resource
 from pydantic import computed_field
@@ -16,13 +15,17 @@ class Cluster(Resource):
  def __repr__(self) -> str:
  return f"Cluster(name='{self.name}', version='{self.version}')"
 
- @computed_field
+ @computed_field # type: ignore
  @property
  def major_minor_version(self) -> float:
- return float(re.match(r"^v?(\d+\.\d+)[^\d]", self.version).groups()[0])
+ match = re.match(r"^v?(\d+\.\d+)[^\d]", self.version)
+ # failed to retrieve, set to a super new version
+ if not match:
+ return 100.0
+ return float(match.groups()[0])
 
  @property
- def db_labels(self) -> Dict[str, str]:
+ def db_labels(self) -> Dict[str, Any]:
  return {
  **self.unique_identifiers,
  "plural": self.plural,

icekube/models/namespace.py

@@ -1,20 +1,21 @@
 from __future__ import annotations
 
-import json
 from functools import cached_property
-from typing import List, Dict, Any
+from typing import Any, Dict, List, cast
 
-from icekube.models.base import RELATIONSHIP, Resource
+from icekube.models.base import Resource
 from pydantic import computed_field
 
 
 class Namespace(Resource):
  supported_api_groups: List[str] = [""]
 
- @computed_field
+ @computed_field # type: ignore
  @cached_property
  def psa_enforce(self) -> str:
- return self.labels.get("pod-security.kubernetes.io/enforce", "privileged")
+ return cast(
+ str, self.labels.get("pod-security.kubernetes.io/enforce", "privileged")
+ )
 
  @property
  def db_labels(self) -> Dict[str, Any]: