Added typing for resource.py

.pre-commit-config.yaml

@@ -41,3 +41,15 @@ repos:
     rev: 37.358.0
     hooks:
       - id: renovate-config-validator
+
+  - repo: https://github.com/gitleaks/gitleaks
+    rev: v8.18.2
+    hooks:
+      - id: gitleaks
+
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.9.0
+    hooks:
+      - id: mypy
+        exclude: ^(tests/|examples/|docs/)
+        additional_dependencies: [types-all]

CONTRIBUTING.md

@@ -10,40 +10,48 @@ To get an overview of the project, read the [README](README.md).
 ## Issues
 
 ### Create a new issue
+
 If you find a problem with the code, [search if an issue already exists](https://github.com/RedHatQE/openshift-python-wrapper/issues).  
 If you open a pull request to fix the problem, an issue will ba automatically created.  
 If a related issue doesn't exist, you can open a new issue using a relevant [issue form](https://github.com/RedHatQE/openshift-python-wrapper/issues/new/choose).
 
 ## Pull requests
+
 To contribute code to the project:
+
 - Fork the project and work on your forked repository
-- Before submitting a new pull request, make sure you have `pre-commit` installed  
+- Before submitting a new pull request, make sure you have `pre-commit` installed
+
 ```bash
 pre-commit install
 ```
+
 - When submitting a pull request, make sure to fill all the required, relevant fields for your PR.  
-Make sure the title is descriptive and short.
+  Make sure the title is descriptive and short.
 - If the fix is needed in a released version, once your pull request is merged, cherry-pick it to the relevant branch(s).  
-Add `/cherry-pick <target branch to cherry-pick to>` to the PR comment.
+  Add `/cherry-pick <target branch to cherry-pick to>` to the PR comment.
 
 ## Adding a new module (resource)
+
 To support working with a new (or missing) resource:
+
 - Add a new file under `ocp_resources`, names as the resource.  
-If the resource name is composed of multiple words, separate them with an underscore.  
-For example: `ImageStream` filename is `image_stream.py`
+  If the resource name is composed of multiple words, separate them with an underscore.  
+  For example: `ImageStream` filename is `image_stream.py`
 - Create a class named as the resource kind.  
-For example: `ImageStream` -> `class ImageStream`
+  For example: `ImageStream` -> `class ImageStream`
 - Inherit from the relevant class.  
-If the resource is cluster-scoped, inherit from `Resource` else from `NamespacedResource`.  
-For example: `class ImageStream(NamespacedResource):`
+  If the resource is cluster-scoped, inherit from `Resource` else from `NamespacedResource`.  
+  For example: `class ImageStream(NamespacedResource):`
 - Add the resource's apiGroup (`apiVersion` prefix) as `api_group`.
-For example: `image.openshift.io`
+  For example: `image.openshift.io`
 - Add a link to the resource's API reference.  
-For example: [ImageStream API reference](https://docs.openshift.com/container-platform/4.11/rest_api/image_apis/imagestream-image-openshift-io-v1.html#imagestream-image-openshift-io-v1)
+  For example: [ImageStream API reference](https://docs.openshift.com/container-platform/4.11/rest_api/image_apis/imagestream-image-openshift-io-v1.html#imagestream-image-openshift-io-v1)
 - Add an `__init__` function.  
-Define all the required arguments that are **required** to instantiate a new resource.  
-Optional parameters may be added as well.  
-For example:
+  Define all the required arguments that are **required** to instantiate a new resource.  
+  Optional parameters may be added as well.  
+  For example:
+
 ```
     def __init__(
         self,
@@ -71,10 +79,12 @@ For example:
         self.tags = tags
         self.lookup_policy = lookup_policy
 ```
+
 - Use `to_dict` to set the values in the resource body. Make sure you call `super().to_dict()` first.  
-For example:
+  For example:
+
 ```
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             self.res.update(
@@ -86,4 +96,5 @@ For example:
                 }
             )
 ```
+
 - Check [imageStreams](ocp_resources/image_stream.py) for reference.

ocp_resources/backup.py

@@ -35,7 +35,7 @@ def __init__(
         self.snapshot_move_data = snapshot_move_data
         self.storage_location = storage_location
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             spec_dict = {}

ocp_resources/catalog_source.py

@@ -33,7 +33,7 @@ def __init__(
         self.publisher = publisher
         self.update_strategy_registry_poll_interval = update_strategy_registry_poll_interval
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not all([self.source_type, self.image, self.display_name, self.publisher]):

ocp_resources/cluster_pool.py

@@ -38,8 +38,6 @@ def __init__(
         self.size = size
 
     def to_dict(self) -> None:
-        self.res: Dict[str, Any]
-
         super().to_dict()
         if not self.yaml_file:
             self.res["spec"] = {}

ocp_resources/cluster_role.py

@@ -20,7 +20,7 @@ def __init__(self, rules=None, **kwargs):
         super().__init__(**kwargs)
         self.rules = rules
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not self.rules:

ocp_resources/cluster_role_binding.py

@@ -26,7 +26,7 @@ def __init__(
         self.cluster_role = cluster_role
         self.subjects = subjects
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not self.cluster_role:

ocp_resources/configmap.py

@@ -20,7 +20,7 @@ def __init__(
         super().__init__(**kwargs)
         self.data = data
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file and self.data:
             self.res.setdefault("data", {}).update(self.data)

ocp_resources/constants.py

@@ -1,3 +1,4 @@
+from typing import Dict, List
 from kubernetes.dynamic.exceptions import (
     ForbiddenError,
     InternalServerError,
@@ -6,7 +7,7 @@
 )
 from urllib3.exceptions import MaxRetryError, ProtocolError
 
-DEFAULT_CLUSTER_RETRY_EXCEPTIONS = {
+DEFAULT_CLUSTER_RETRY_EXCEPTIONS: Dict[type[Exception], List[str]] = {
     MaxRetryError: [],
     ConnectionAbortedError: [],
     ConnectionResetError: [],
@@ -19,11 +20,11 @@
     ServerTimeoutError: [],
     ForbiddenError: ["context deadline exceeded"],
 }
-PROTOCOL_ERROR_EXCEPTION_DICT = {ProtocolError: []}
-NOT_FOUND_ERROR_EXCEPTION_DICT = {NotFoundError: []}
+PROTOCOL_ERROR_EXCEPTION_DICT: Dict[type[Exception], List[str]] = {ProtocolError: []}
+NOT_FOUND_ERROR_EXCEPTION_DICT: Dict[type[Exception], List[str]] = {NotFoundError: []}
 
-TIMEOUT_10SEC = 10
-TIMEOUT_1MINUTE = 60
-TIMEOUT_2MINUTES = 2 * 60
-TIMEOUT_4MINUTES = 4 * 60
-TIMEOUT_10MINUTES = 10 * 60
+TIMEOUT_10SEC: int = 10
+TIMEOUT_1MINUTE: int = 60
+TIMEOUT_2MINUTES: int = 2 * 60
+TIMEOUT_4MINUTES: int = 4 * 60
+TIMEOUT_10MINUTES: int = 10 * 60

ocp_resources/controller_revision.py

@@ -26,7 +26,7 @@ def __init__(
         self.revision_object = revision_object
         self.revision = revision
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not self.revision:

ocp_resources/cron_job.py

@@ -44,7 +44,7 @@ def __init__(
         self.failed_jobs_history_limit = failed_jobs_history_limit
         self.starting_deadline_seconds = starting_deadline_seconds
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not (self.job_template and self.schedule):

ocp_resources/csi_storage_capacity.py

@@ -32,7 +32,7 @@ def __init__(
         self.storage_class_name = storage_class_name
         self.maximum_volume_size = maximum_volume_size
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not self.storage_class_name:

ocp_resources/data_import_cron.py

@@ -53,7 +53,7 @@ def __init__(
         self.imports_to_keep = imports_to_keep
         self.bind_immediate_annotation = bind_immediate_annotation
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if self.image_stream and self.url:

ocp_resources/data_source.py

@@ -22,7 +22,7 @@ def __init__(self, source=None, **kwargs):
         super().__init__(**kwargs)
         self._source = source
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not self._source:

ocp_resources/datavolume.py

@@ -153,7 +153,7 @@ def __init__(
         self.api_name = api_name
         self.delete_after_completion = delete_after_completion
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             self.res.update({

ocp_resources/endpoint_slice.py

@@ -51,7 +51,7 @@ def __init__(
         self.endpoints = endpoints
         self.ports = ports
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not (self.address_type and self.endpoints):

ocp_resources/endpoints.py

@@ -48,11 +48,12 @@ def __init__(
         self.addresses = addresses
         self.ports = ports
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not (self.addresses and self.ports):
-                raise MissingRequiredArgumentError(argumens="'addresses' and 'ports")
+                raise MissingRequiredArgumentError(argument="'addresses' and 'ports")
+
             self.res.update({
                 "subsets": {
                     "addresses": self.addresses,

ocp_resources/group.py

@@ -12,7 +12,7 @@ def __init__(
         super().__init__(**kwargs)
         self.users = users
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if (not self.yaml_file) and self.users:
             self.res["users"] = self.users

ocp_resources/hook.py

@@ -39,7 +39,7 @@ def __init__(
         self.image = image
         self.playbook = playbook
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             self.res.update({

ocp_resources/host.py

@@ -43,7 +43,7 @@ def __init__(
         self.secret_namespace = secret_namespace or self.namespace
         self.condition_message_ready = self.ConditionMessage.HOST_READY
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             self.res.update({

ocp_resources/hostpath_provisioner.py

@@ -35,7 +35,7 @@ def __init__(
         self.path = path
         self.image_pull_policy = image_pull_policy
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             spec = self.res.setdefault("spec", {})

ocp_resources/hyperconverged.py

@@ -29,7 +29,7 @@ def __init__(
         self.infra = infra
         self.workloads = workloads
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if self.infra:

ocp_resources/image_content_source_policy.py

@@ -23,7 +23,7 @@ def __init__(self, repository_digest_mirrors=None, **kwargs):
         self.repository_digest_mirrors = repository_digest_mirrors
         super().__init__(**kwargs)
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not self.repository_digest_mirrors:

ocp_resources/image_digest_mirror_set.py

@@ -23,7 +23,7 @@ def __init__(self, image_digest_mirrors=None, **kwargs):
         self.image_digest_mirrors = image_digest_mirrors
         super().__init__(**kwargs)
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not self.image_digest_mirrors:

ocp_resources/image_stream.py

@@ -36,7 +36,7 @@ def __init__(
         self.tags = tags
         self.lookup_policy = lookup_policy
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             self.res.update({

ocp_resources/ip_address_pool.py

@@ -39,7 +39,7 @@ def __init__(
         self.auto_assign = auto_assign
         self.avoid_buggy_ips = avoid_buggy_ips
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not self.addresses:

ocp_resources/job.py

@@ -63,7 +63,7 @@ def __init__(
         self.containers = containers
         self.background_propagation_policy = background_propagation_policy
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             self.res.setdefault("spec", {})

ocp_resources/kube_descheduler.py

@@ -48,7 +48,7 @@ def __init__(
         self.mode = mode
         self.operator_log_level = operator_log_level
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             self.res.update({

ocp_resources/l2_advertisement.py

@@ -40,7 +40,7 @@ def __init__(
         self.ip_address_pools = ip_address_pools
         self.ip_address_pools_selectors = ip_address_pools_selectors
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             self.res["spec"] = {}

ocp_resources/lease.py

@@ -57,7 +57,7 @@ def __init__(
         self.acquire_time = acquire_time
         self.renew_time = renew_time
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if self.acquire_time:

ocp_resources/limit_range.py

@@ -21,7 +21,7 @@ def __init__(
         super().__init__(**kwargs)
         self.limits = limits
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             if not self.limits:

ocp_resources/machine_config_pool.py

@@ -55,7 +55,7 @@ def __init__(
         self.max_unavailable = max_unavailable
         self.paused = paused
 
-    def to_dict(self):
+    def to_dict(self) -> None:
         super().to_dict()
         if not self.yaml_file:
             self.res.update(