Merge pull request #9 from hactar-is/typing-fix

Fixes uses of type :str = None to be :Optional[str] = None

Author: nimish-ks

src/phase/utils/phase_io.py

@@ -1,23 +1,23 @@
-import requests
-from typing import Tuple
-from typing import List, Dict
 from dataclasses import dataclass
+from typing import Dict, List, Optional, Tuple
+
+import requests
+from nacl.bindings import (
+    crypto_kx_server_session_keys,
+)
+
+from .const import pss_service_pattern, pss_user_pattern
+from .crypto import CryptoUtils
+from .misc import phase_get_context, tag_matches
 from .network import (
-    fetch_phase_user,
+    create_phase_secrets,
+    delete_phase_secrets,
     fetch_app_key,
-    fetch_wrapped_key_share,
     fetch_phase_secrets,
-    create_phase_secrets,
+    fetch_phase_user,
+    fetch_wrapped_key_share,
     update_phase_secrets,
-    delete_phase_secrets
-)
-from nacl.bindings import (
-    crypto_kx_server_session_keys, 
 )
-from .crypto import CryptoUtils
-from .const import __ph_version__, pss_user_pattern, pss_service_pattern
-from .misc import phase_get_context, normalize_tag, tag_matches
-from .secret_referencing import resolve_all_secrets
 
 
 @dataclass
@@ -80,8 +80,8 @@ def auth(self):
 
             return "Success"
 
-        except ValueError as err:
-            raise ValueError(f"Invalid Phase credentials")
+        except ValueError:
+            raise ValueError("Invalid Phase credentials")
 
 
     def init(self):
@@ -90,12 +90,12 @@ def init(self):
         # Ensure the response is OK
         if response.status_code != 200:
             raise ValueError(f"Request failed with status code {response.status_code}: {response.text}")
-        
+
         # Parse and return the JSON content
         return response.json()
 
 
-    def create(self, key_value_pairs: List[Tuple[str, str]], env_name: str, app_name: str, path: str = '/', override_value: str = None) -> requests.Response:
+    def create(self, key_value_pairs: List[Tuple[str, str]], env_name: str, app_name: str, path: str = '/', override_value: Optional[str] = None) -> requests.Response:
         """
         Create secrets in Phase KMS with support for specifying a path and overrides.
 
@@ -150,7 +150,7 @@ def create(self, key_value_pairs: List[Tuple[str, str]], env_name: str, app_name
         return create_phase_secrets(self._token_type, self._app_secret.app_token, env_id, secrets, self._api_host)
 
 
-    def get(self, env_name: str, keys: List[str] = None, app_name: str = None, tag: str = None, path: str = '') -> List[Dict]:
+    def get(self, env_name: str, keys: List[str] = None, app_name: Optional[str] = None, tag: Optional[str] = None, path: str = '') -> List[Dict]:
         """
         Get secrets from Phase KMS based on key and environment, with support for personal overrides,
         optional tag matching, decrypting comments, and now including path support and key digest optimization.
@@ -165,7 +165,7 @@ def get(self, env_name: str, keys: List[str] = None, app_name: str = None, tag:
         Returns:
             List[Dict]: A list of dictionaries for all secrets in the environment that match the criteria, including their paths.
         """
-        
+
         user_response = fetch_phase_user(self._token_type, self._app_secret.app_token, self._api_host)
         if user_response.status_code != 200:
             raise ValueError(f"Request failed with status code {user_response.status_code}: {user_response.text}")
@@ -221,7 +221,7 @@ def get(self, env_name: str, keys: List[str] = None, app_name: str = None, tag:
                 "comment": decrypted_comment,
                 "path": secret.get("path", "/"),
                 "application": app_name,
-                "environment": env_name 
+                "environment": env_name
             }
 
             # Only add the secret to results if the requested keys are not specified or the decrypted key is one of the requested keys.
@@ -231,10 +231,10 @@ def get(self, env_name: str, keys: List[str] = None, app_name: str = None, tag:
         return results
 
 
-    def update(self, env_name: str, key: str, value: str = None, app_name: str = None, source_path: str = '', destination_path: str = None, override: bool = False, toggle_override: bool = False) -> str:
+    def update(self, env_name: str, key: str, value: Optional[str] = None, app_name: Optional[str] = None, source_path: str = '', destination_path: Optional[str] = None, override: bool = False, toggle_override: bool = False) -> str:
         """
         Update a secret in Phase KMS based on key and environment, with support for source and destination paths.
-        
+
         Args:
             env_name (str): The name (or partial name) of the desired environment.
             key (str): The key for which to update the secret value.
@@ -244,11 +244,11 @@ def update(self, env_name: str, key: str, value: str = None, app_name: str = Non
             destination_path (str, optional): The new path for the secret, if changing its location. If not provided, the path is not updated.
             override (bool, optional): Whether to update an overridden secret value. Defaults to False.
             toggle_override (bool, optional): Whether to toggle the override state between active and inactive. Defaults to False.
-                
+
         Returns:
             str: A message indicating the outcome of the update operation.
         """
-        
+
         user_response = fetch_phase_user(self._token_type, self._app_secret.app_token, self._api_host)
         if user_response.status_code != 200:
             raise ValueError(f"Request failed with status code {user_response.status_code}: {user_response.text}")
@@ -301,10 +301,10 @@ def update(self, env_name: str, key: str, value: str = None, app_name: str = Non
             # This prevents toggling an override on a secret that doesn't have one.
             if "override" not in matching_secret or matching_secret["override"] is None:
                 raise OverrideNotFoundException(key)
-            
+
             # Retrieve the current override state. If the override is not active, it defaults to False.
             current_override_state = matching_secret["override"].get("is_active", False)
-            
+
             # Prepare the payload to update the override status. The value of the override remains unchanged,
             # but the isActive status is toggled.
             secret_update_payload["override"] = {
@@ -338,20 +338,20 @@ def update(self, env_name: str, key: str, value: str = None, app_name: str = Non
             return f"Error: Failed to update secret. HTTP Status Code: {response.status_code}"
 
 
-    def delete(self, env_name: str, keys_to_delete: List[str], app_name: str = None, path: str = None) -> List[str]:
+    def delete(self, env_name: str, keys_to_delete: List[str], app_name: Optional[str] = None, path: Optional[str] = None) -> List[str]:
         """
         Delete secrets in Phase KMS based on keys and environment, with optional path support.
-        
+
         Args:
             env_name (str): The name (or partial name) of the desired environment.
             keys_to_delete (List[str]): The keys for which to delete the secrets.
             app_name (str, optional): The name of the desired application.
             path (str, optional): The path within which to delete the secrets. If specified, only deletes secrets within this path.
-                
+
         Returns:
             List[str]: A list of keys that were not found and could not be deleted.
         """
-        
+
         user_response = fetch_phase_user(self._token_type, self._app_secret.app_token, self._api_host)
         if user_response.status_code != 200:
             raise ValueError(f"Request failed with status code {user_response.status_code}: {user_response.text}")
@@ -372,7 +372,7 @@ def delete(self, env_name: str, keys_to_delete: List[str], app_name: str = None,
         keys_not_found = []
         secrets_response = fetch_phase_secrets(self._token_type, self._app_secret.app_token, env_id, self._api_host, path=path)
         secrets_data = secrets_response.json()
-            
+
         for key in keys_to_delete:
             found = False
             for secret in secrets_data:
@@ -388,9 +388,9 @@ def delete(self, env_name: str, keys_to_delete: List[str], app_name: str = None,
 
         if secret_ids_to_delete:
             delete_phase_secrets(self._token_type, self._app_secret.app_token, env_id, secret_ids_to_delete, self._api_host)
-            
+
         return keys_not_found
-    
+
 
     def decrypt(self, phase_ciphertext) -> str | None:
         """