Merge pull request #10 from LamaAni/fix_packages

Fix packages

Author: LamaAni

Pipfile

@@ -7,10 +7,12 @@ verify_ssl = true
 pytest = "*"
 black = "*"
 flake8 = "*"
+apache-ariflow = "==1.10.9"
+airflow-kubernetes-job-operator = {path = "."}
 
 [packages]
 pyyaml = ">=5.0"
-kubernetes = ">=9.0.0"
+kubernetes = ">=8.0.1"
 urllib3 = ">=1.25.0"
 zthreading = ">=0.1.13"
 python-dateutil = "*"

airflow_kubernetes_job_operator/kube_api/client.py

@@ -66,6 +66,7 @@ def __init__(
         auto_reconnect: bool = False,
         auto_reconnect_max_attempts: int = DEFAULT_AUTO_RECONNECT_MAX_ATTEMPTS,
         auto_reconnect_wait_between_attempts: float = DEFAULT_AUTO_RECONNECT_WAIT_BETWEEN_ATTEMPTS,
+        always_throw_on_first_api_call: bool = True,
     ):
         assert use_asyncio is not True, NotImplementedError("AsyncIO not yet implemented.")
         super().__init__(
@@ -87,6 +88,7 @@ def __init__(
         self.auto_reconnect = auto_reconnect
         self.auto_reconnect_max_attempts = auto_reconnect_max_attempts
         self.auto_reconnect_wait_between_attempts = auto_reconnect_wait_between_attempts
+        self.always_throw_on_first_api_call = always_throw_on_first_api_call
 
         # these event are object specific
         self.query_started_event_name = f"{self.query_started_event_name} {id(self)}"
@@ -257,9 +259,10 @@ def can_reconnect(message=None):
                         exception_message = (
                             f"{ex.reason}, {exception_details.get('reason')}: {exception_details.get('message')}"
                         )
-                    if can_reconnect(True, exception_message):
+                    force_throw = self.always_throw_on_first_api_call and total_reconnects == 0
+                    if not force_throw and can_reconnect(exception_message):
                         continue
-                    if self.query_running and not self._is_being_stopped:
+                    if force_throw or self.query_running and not self._is_being_stopped:
                         raise KubeApiClientException(exception_message, inner_exception=ex)
                 else:
                     raise ex
@@ -405,7 +408,7 @@ def load_kubernetes_configuration_from_file(
         )
 
         configuration = kube_config.Configuration()
-
+        configuration.filepath = None
         if is_in_cluster and config_file is None:
             # load from cluster.
             loader = incluster_config.InClusterConfigLoader(
@@ -423,7 +426,8 @@ def load_kubernetes_configuration_from_file(
                     DEFAULT_KUBE_CONFIG_LOCATIONS,
                 )
                 for loc in config_possible_locations:
-                    if os.path.isfile(expanduser(loc)):
+                    loc = expanduser(loc)
+                    if os.path.isfile(loc):
                         config_file = loc
 
             assert config_file is not None, "Kubernetes config file not provided and default config could not be found."

airflow_kubernetes_job_operator/kube_api/queries.py

@@ -168,6 +168,7 @@ def __init__(
                 "watch": watch,
             },
             auto_reconnect=watch,
+            always_throw_on_first_api_call=name is None,
         )
         self.kind = kind
         self.namespace = namespace

setup.py

@@ -43,7 +43,7 @@ def get_version():
     license="LICENSE",
     install_requires=[
         "PyYAML>=5.0",
-        "kubernetes>=9.0.0",
+        "kubernetes>=8.0.1",
         "urllib3>=1.25.0",
         "zthreading>=0.1.13",
         "python-dateutil>=2.8.1",