Merge pull request #8 from phasehq/feat--app-id

Feat: added app-id & service account support

Author: nimish-ks

README.md

@@ -111,21 +111,6 @@ result = phase.delete_secret(delete_options)
 print(f"Delete result: {result}")
 ```
 
-### Resolve Secret References
-
-Resolve references in secret values:
-
-```python
-get_options = GetAllSecretsOptions(
-    env_name="Development",
-    app_name="Your App Name"
-)
-secrets = phase.get_all_secrets(get_options)
-resolved_secrets = phase.resolve_references(secrets, "Development", "Your App Name")
-for secret in resolved_secrets:
-    print(f"Key: {secret.key}, Resolved Value: {secret.value}")
-```
-
 ## Error Handling
 
 The SDK methods may raise exceptions for various error conditions. It's recommended to wrap SDK calls in try-except blocks to handle potential errors:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "phase_dev"
-version = "2.0.1"
+version = "2.1.0"
 description = "Python SDK for Phase secrets manager"
 readme = "README.md"
 requires-python = ">=3.10"

src/phase/phase.py

@@ -6,43 +6,68 @@
 @dataclass
 class GetSecretOptions:
     env_name: str
-    app_name: str
+    app_name: Optional[str] = None
+    app_id: Optional[str] = None
     key_to_find: Optional[str] = None
     tag: Optional[str] = None
     secret_path: str = "/"
 
+    def __post_init__(self):
+        if not self.app_name and not self.app_id:
+            raise ValueError("Either app_name or app_id must be provided")
+
 @dataclass
 class GetAllSecretsOptions:
     env_name: str
-    app_name: str
+    app_name: Optional[str] = None
+    app_id: Optional[str] = None
     tag: Optional[str] = None
     secret_path: str = "/"
 
+    def __post_init__(self):
+        if not self.app_name and not self.app_id:
+            raise ValueError("Either app_name or app_id must be provided")
+
 @dataclass
 class CreateSecretsOptions:
     env_name: str
-    app_name: str
     key_value_pairs: List[Dict[str, str]]
+    app_name: Optional[str] = None
+    app_id: Optional[str] = None
     secret_path: str = "/"
 
+    def __post_init__(self):
+        if not self.app_name and not self.app_id:
+            raise ValueError("Either app_name or app_id must be provided")
+
 @dataclass
 class UpdateSecretOptions:
     env_name: str
-    app_name: str
     key: str
     value: Optional[str] = None
+    app_name: Optional[str] = None
+    app_id: Optional[str] = None
     secret_path: str = "/"
     destination_path: Optional[str] = None
     override: bool = False
     toggle_override: bool = False
 
+    def __post_init__(self):
+        if not self.app_name and not self.app_id:
+            raise ValueError("Either app_name or app_id must be provided")
+
 @dataclass
 class DeleteSecretOptions:
     env_name: str
-    app_name: str
     key_to_delete: str
+    app_name: Optional[str] = None
+    app_id: Optional[str] = None
     secret_path: str = "/"
 
+    def __post_init__(self):
+        if not self.app_name and not self.app_id:
+            raise ValueError("Either app_name or app_id must be provided")
+
 @dataclass
 class PhaseSecret:
     key: str
@@ -51,49 +76,127 @@ class PhaseSecret:
     path: str = "/"
     tags: List[str] = field(default_factory=list)
     overridden: bool = False
+    application: Optional[str] = None
+    environment: Optional[str] = None
 
 class Phase:
     def __init__(self, init=True, pss=None, host=None):
         self._phase_io = PhaseIO(init=init, pss=pss, host=host)
 
+    def _resolve_secret_values(self, secrets: List[PhaseSecret], env_name: str, app_name: str) -> List[PhaseSecret]:
+        """
+        Utility function to resolve secret references within secret values.
+        
+        Args:
+            secrets (List[PhaseSecret]): List of secrets to process
+            env_name (str): Environment name for secret resolution
+            app_name (str): Application name for secret resolution
+            
+        Returns:
+            List[PhaseSecret]: List of secrets with resolved values
+        """
+        # Convert PhaseSecret objects to dict format expected by resolve_all_secrets
+        all_secrets = [
+            {
+                'environment': secret.environment or env_name,
+                'path': secret.path,
+                'key': secret.key,
+                'value': secret.value
+            }
+            for secret in secrets
+        ]
+        
+        # Create new list of secrets with resolved values
+        resolved_secrets = []
+        for secret in secrets:
+            resolved_value = resolve_all_secrets(
+                value=secret.value,
+                all_secrets=all_secrets,
+                phase=self._phase_io,
+                current_application_name=secret.application or app_name,
+                current_env_name=secret.environment or env_name
+            )
+            
+            resolved_secrets.append(PhaseSecret(
+                key=secret.key,
+                value=resolved_value,
+                comment=secret.comment,
+                path=secret.path,
+                tags=secret.tags,
+                overridden=secret.overridden,
+                application=secret.application,
+                environment=secret.environment
+            ))
+            
+        return resolved_secrets
+
     def get_secret(self, options: GetSecretOptions) -> Optional[PhaseSecret]:
         secrets = self._phase_io.get(
             env_name=options.env_name,
             keys=[options.key_to_find] if options.key_to_find else None,
             app_name=options.app_name,
+            app_id=options.app_id,
             tag=options.tag,
             path=options.secret_path
         )
         if secrets:
             secret = secrets[0]
-            return PhaseSecret(
+            phase_secret = PhaseSecret(
                 key=secret['key'],
                 value=secret['value'],
                 comment=secret.get('comment', ''),
                 path=secret.get('path', '/'),
                 tags=secret.get('tags', []),
-                overridden=secret.get('overridden', False)
+                overridden=secret.get('overridden', False),
+                application=secret.get('application'),
+                environment=secret.get('environment')
+            )
+
+            # Resolve any secret references in the value
+            resolved_secrets = self._resolve_secret_values(
+                [phase_secret], 
+                options.env_name,
+                secret.get('application', options.app_name)
             )
+            
+            return resolved_secrets[0] if resolved_secrets else None
         return None
 
     def get_all_secrets(self, options: GetAllSecretsOptions) -> List[PhaseSecret]:
         secrets = self._phase_io.get(
             env_name=options.env_name,
             app_name=options.app_name,
+            app_id=options.app_id,
             tag=options.tag,
             path=options.secret_path
         )
-        return [
+        
+        if not secrets:
+            return []
+
+        # Get the application name from the first secret
+        app_name = secrets[0].get('application', options.app_name)
+        
+        phase_secrets = [
             PhaseSecret(
                 key=secret['key'],
                 value=secret['value'],
                 comment=secret.get('comment', ''),
                 path=secret.get('path', '/'),
                 tags=secret.get('tags', []),
-                overridden=secret.get('overridden', False)
+                overridden=secret.get('overridden', False),
+                application=secret.get('application'),
+                environment=secret.get('environment')
             )
             for secret in secrets
         ]
+        
+        # Resolve any secret references in the values
+        return self._resolve_secret_values(
+            phase_secrets,
+            options.env_name,
+            app_name
+        )
 
     def create_secrets(self, options: CreateSecretsOptions) -> str:
         # Convert the list of dictionaries to a list of tuples
@@ -103,6 +206,7 @@ def create_secrets(self, options: CreateSecretsOptions) -> str:
             key_value_pairs=key_value_tuples,
             env_name=options.env_name,
             app_name=options.app_name,
+            app_id=options.app_id,
             path=options.secret_path
         )
         return "Success" if response.status_code == 200 else f"Error: {response.status_code}"
@@ -113,6 +217,7 @@ def update_secret(self, options: UpdateSecretOptions) -> str:
             key=options.key,
             value=options.value,
             app_name=options.app_name,
+            app_id=options.app_id,
             source_path=options.secret_path,
             destination_path=options.destination_path,
             override=options.override,
@@ -124,29 +229,6 @@ def delete_secret(self, options: DeleteSecretOptions) -> List[str]:
             env_name=options.env_name,
             keys_to_delete=[options.key_to_delete],
             app_name=options.app_name,
+            app_id=options.app_id,
             path=options.secret_path
         )
-
-    def resolve_references(self, secrets: List[PhaseSecret], env_name: str, app_name: str) -> List[PhaseSecret]:
-        all_secrets = [
-            {
-                'environment': env_name,
-                'application': app_name,
-                'key': secret.key,
-                'value': secret.value,
-                'path': secret.path
-            }
-            for secret in secrets
-        ]
-        
-        for secret in secrets:
-            resolved_value = resolve_all_secrets(
-                secret.value, 
-                all_secrets, 
-                self._phase_io, 
-                app_name, 
-                env_name
-            )
-            secret.value = resolved_value
-        
-        return secrets

src/phase/utils/const.py

@@ -1,7 +1,7 @@
 import os
 import re
 
-__version__ = "2.0.1"
+__version__ = "2.1.0"
 __ph_version__ = "v1"
 
 

src/phase/utils/misc.py

@@ -42,44 +42,43 @@ def get_default_user_token() -> str:
     raise ValueError("Default user not found in the config file.")
 
 
-def phase_get_context(user_data, app_name=None, env_name=None):
+def phase_get_context(user_data, app_name=None, env_name=None, app_id=None):
     """
     Get the context (ID, name, and publicKey) for a specified application and environment or the default application and environment.
 
     Parameters:
     - user_data (dict): The user data from the API response.
     - app_name (str, optional): The name (or partial name) of the desired application.
     - env_name (str, optional): The name (or partial name) of the desired environment.
+    - app_id (str, optional): The explicit application ID to use. Takes precedence over app_name if both are provided.
 
     Returns:
     - tuple: A tuple containing the application's name, application's ID, environment's name, environment's ID, and publicKey.
 
     Raises:
     - ValueError: If no matching application or environment is found.
     """
-
-    # 2. If env_name isn't explicitly provided, use the default
+    # 1. Set default environment name
     default_env_name = "Development"
-    app_id = None
     env_name = env_name or default_env_name
 
-    # 3. Match the application using app_id or find the best match for partial app_name
+    # 2. Match the application using app_id first, then fall back to app_name if app_id is not provided
     try:
-        if app_name:
+        if app_id:  # app_id takes precedence
+            application = next((app for app in user_data["apps"] if app["id"] == app_id), None)
+            if not application:
+                raise ValueError(f"🔍 No application found with ID: '{app_id}'.")
+        elif app_name:  # only check app_name if app_id is not provided
             matching_apps = [app for app in user_data["apps"] if app_name.lower() in app["name"].lower()]
             if not matching_apps:
                 raise ValueError(f"🔍 No application found with the name '{app_name}'.")
             # Sort matching applications by the length of their names, shorter names are likely to be more specific matches
             matching_apps.sort(key=lambda app: len(app["name"]))
             application = matching_apps[0]
-        elif app_id:
-            application = next((app for app in user_data["apps"] if app["id"] == app_id), None)
-            if not application:
-                raise ValueError(f"🔍 No application found with the name '{app_name_from_config}' and ID: '{app_id}'.")
         else:
-            raise ValueError("🤔 No application context provided. Please run 'phase init' or pass the '--app' flag followed by your application name.")
+            raise ValueError("🤔 No application context provided. Please provide either app_name or app_id.")
 
-        # 4. Attempt to match environment with the exact name or a name that contains the env_name string
+        # 3. Attempt to match environment with the exact name or a name that contains the env_name string
         environment = next((env for env in application["environment_keys"] if env_name.lower() in env["environment"]["name"].lower()), None)
 
         if not environment:

src/phase/utils/network.py

@@ -77,7 +77,7 @@ def construct_http_headers(token_type: str, app_token: str) -> Dict[str, str]:
         Dict[str, str]: The common headers including User-Agent.
     """
     return {
-        "Authorization": f"Bearer {token_type.capitalize()} {app_token}",
+        "Authorization": f"Bearer {token_type} {app_token}",
         "User-Agent": get_user_agent()
     }