Skip to content

Commit

Permalink
fix(typing): class attributes (#52)
Browse files Browse the repository at this point in the history
  • Loading branch information
maxime1907 authored Apr 19, 2024
1 parent cddebbe commit 793399b
Show file tree
Hide file tree
Showing 5 changed files with 40 additions and 24 deletions.
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
# wiremind-kubernetes

## v7.4.3 (2024-04-19)
### Fix
- correct class attributes typing

## v7.4.2 (2024-04-19)
### Fix
- kubernetes_helper: correct typing for `use_kubeconfig`
Expand Down
2 changes: 1 addition & 1 deletion VERSION
Original file line number Diff line number Diff line change
@@ -1 +1 @@
7.4.2
7.4.3
2 changes: 2 additions & 0 deletions src/wiremind_kubernetes/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@ class WiremindKubernetesException(Exception):
Base wiremind-kubernetes Exception.
"""

message: str

def __init__(self, message: Optional[str] = None):
super().__init__()
if message:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,15 +10,19 @@ class ClientWithArguments:
Currently add dry_run support for write functions and pretty to all.
"""

client: Any
read_additional_arguments: Dict[str, Any]
additional_arguments: Dict[str, Any]

def __init__(self, client: Any, dry_run: bool = False, pretty: bool = True):
self.client = client() # like kubernetes.client.CoreV1Api
self.read_additional_arguments: Dict[str, Any] = {}
self.read_additional_arguments = {}
# Only add it when its true because we set pretty client wide,
# read_cluster_custom_object which accepts it will not have it set, but it's ok for now.
if pretty:
self.read_additional_arguments["pretty"] = pretty
# Every request, either read or write, will have those arguments added
self.additional_arguments: Dict[str, Any] = self.read_additional_arguments.copy()
self.additional_arguments = self.read_additional_arguments.copy()
if dry_run:
# Dry run, in kube API, is not true or false, but either dry_run: All or not defined.
self.additional_arguments["dry_run"] = "All"
Expand Down
48 changes: 27 additions & 21 deletions src/wiremind_kubernetes/kubernetes_helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,19 @@ class KubernetesHelper:
A simple helper for Kubernetes manipulation.
"""

SCALE_DOWN_MAX_WAIT_TIME = 3600
SCALE_DOWN_MAX_WAIT_TIME: int = 3600

client_corev1_api: kubernetes.client.CoreV1Api
client_appsv1_api: kubernetes.client.AppsV1Api
client_batchv1_api: kubernetes.client.BatchV1Api
client_autoscalingv1_api: kubernetes.client.AutoscalingV1Api
client_custom_objects_api: kubernetes.client.CustomObjectsApi
client_rbac_authorization_v1_api: kubernetes.client.RbacAuthorizationV1Api
client_networking_v1_api: kubernetes.client.NetworkingV1Api
client_storage_v1_api: kubernetes.client.StorageV1Api

dry_run: bool
pretty: bool

def __init__(
self,
Expand All @@ -52,27 +64,17 @@ def __init__(
"""
if should_load_kubernetes_config:
load_kubernetes_config(use_kubeconfig=use_kubeconfig, context=context)
self.client_corev1_api: kubernetes.client.CoreV1Api = CoreV1ApiWithArguments(dry_run=dry_run, pretty=pretty)
self.client_appsv1_api: kubernetes.client.AppsV1Api = AppV1ApiWithArguments(dry_run=dry_run, pretty=pretty)
self.client_batchv1_api: kubernetes.client.BatchV1Api = BatchV1ApiWithArguments(dry_run=dry_run, pretty=pretty)
self.client_autoscalingv1_api: kubernetes.client.AutoscalingV1Api = AutoscalingV1ApiWithArguments(
dry_run=dry_run, pretty=pretty
)
self.client_custom_objects_api: kubernetes.client.CustomObjectsApi = CustomObjectsApiWithArguments(
dry_run=dry_run, pretty=pretty
)
self.client_rbac_authorization_v1_api: kubernetes.client.RbacAuthorizationV1Api = (
RbacAuthorizationV1ApiWithArguments(dry_run=dry_run, pretty=pretty)
)
self.client_networking_v1_api: kubernetes.client.NetworkingV1Api = NetworkingV1ApiWithArguments(
dry_run=dry_run, pretty=pretty
)
self.client_storage_v1_api: kubernetes.client.StorageV1Api = StorageV1ApiWithArguments(
dry_run=dry_run, pretty=pretty
)
self.client_corev1_api = CoreV1ApiWithArguments(dry_run=dry_run, pretty=pretty)
self.client_appsv1_api = AppV1ApiWithArguments(dry_run=dry_run, pretty=pretty)
self.client_batchv1_api = BatchV1ApiWithArguments(dry_run=dry_run, pretty=pretty)
self.client_autoscalingv1_api = AutoscalingV1ApiWithArguments(dry_run=dry_run, pretty=pretty)
self.client_custom_objects_api = CustomObjectsApiWithArguments(dry_run=dry_run, pretty=pretty)
self.client_rbac_authorization_v1_api = RbacAuthorizationV1ApiWithArguments(dry_run=dry_run, pretty=pretty)
self.client_networking_v1_api = NetworkingV1ApiWithArguments(dry_run=dry_run, pretty=pretty)
self.client_storage_v1_api = StorageV1ApiWithArguments(dry_run=dry_run, pretty=pretty)

self.dry_run: bool = dry_run
self.pretty: bool = pretty
self.dry_run = dry_run
self.pretty = pretty


def _get_namespace_from_kube() -> str:
Expand All @@ -84,6 +86,8 @@ class NamespacedKubernetesHelper(KubernetesHelper):
A simple helper for Kubernetes manipulation.
"""

namespace: str

def __init__(
self,
use_kubeconfig: Optional[bool] = False,
Expand Down Expand Up @@ -254,6 +258,8 @@ class KubernetesDeploymentManager(NamespacedKubernetesHelper):
a.start_pods()
"""

release_name: str

def __init__(self, release_name: str, **kwargs: Any):
self.release_name = release_name
super().__init__(**kwargs)
Expand Down

0 comments on commit 793399b

Please sign in to comment.