Pavel Baksy
f4f33324dd
Each environment column header now has a copy button that creates a duplicate with a unique name (e.g. "Production (1)"). Non-sensitive variables are copied directly; sensitive variables are migrated through the keyring asynchronously.
749 lines
28 KiB
Python
749 lines
28 KiB
Python
# project_manager.py
|
|
#
|
|
# Copyright 2025 Pavel Baksy
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
#
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
import json
|
|
import uuid
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import List, Optional, Callable
|
|
from datetime import datetime, timezone
|
|
import gi
|
|
gi.require_version('GLib', '2.0')
|
|
from gi.repository import GLib
|
|
from .models import Project, SavedRequest, HttpRequest, Environment
|
|
from .secret_manager import get_secret_manager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ProjectManager:
|
|
"""Manages project and saved request persistence."""
|
|
|
|
def __init__(self):
|
|
# Use XDG data directory (works for both Flatpak and native)
|
|
# Flatpak: ~/.var/app/cz.bugsy.roster/data/cz.bugsy.roster
|
|
# Native: ~/.local/share/cz.bugsy.roster
|
|
self.data_dir = Path(GLib.get_user_data_dir()) / 'cz.bugsy.roster'
|
|
self.projects_file = self.data_dir / 'requests.json'
|
|
self._ensure_data_dir()
|
|
|
|
# In-memory cache for projects to reduce disk I/O
|
|
self._projects_cache: Optional[List[Project]] = None
|
|
|
|
def _ensure_data_dir(self):
|
|
"""Create data directory if it doesn't exist."""
|
|
self.data_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def load_projects(self, force_reload: bool = False) -> List[Project]:
|
|
"""
|
|
Load projects from JSON file with in-memory caching.
|
|
|
|
Args:
|
|
force_reload: If True, bypass cache and reload from disk
|
|
|
|
Returns:
|
|
List of Project objects
|
|
"""
|
|
# Return cached data if available and not forcing reload
|
|
if self._projects_cache is not None and not force_reload:
|
|
return self._projects_cache
|
|
|
|
# Load from disk
|
|
if not self.projects_file.exists():
|
|
self._projects_cache = []
|
|
return []
|
|
|
|
try:
|
|
with open(self.projects_file, 'r') as f:
|
|
data = json.load(f)
|
|
self._projects_cache = [Project.from_dict(p) for p in data.get('projects', [])]
|
|
return self._projects_cache
|
|
except Exception as e:
|
|
logger.error(f"Error loading projects: {e}")
|
|
self._projects_cache = []
|
|
return []
|
|
|
|
def save_projects(self, projects: List[Project]):
|
|
"""Save projects to JSON file and update cache."""
|
|
try:
|
|
data = {
|
|
'version': 1,
|
|
'projects': [p.to_dict() for p in projects]
|
|
}
|
|
with open(self.projects_file, 'w') as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
# Update cache with the saved data
|
|
self._projects_cache = projects
|
|
except Exception as e:
|
|
logger.error(f"Error saving projects: {e}")
|
|
|
|
def add_project(self, name: str) -> Project:
|
|
"""Create new project with default environment."""
|
|
projects = self.load_projects()
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
|
|
# Create default environment
|
|
default_env = Environment(
|
|
id=str(uuid.uuid4()),
|
|
name="Default",
|
|
variables={},
|
|
created_at=now
|
|
)
|
|
|
|
project = Project(
|
|
id=str(uuid.uuid4()),
|
|
name=name,
|
|
requests=[],
|
|
created_at=now,
|
|
variable_names=[],
|
|
environments=[default_env]
|
|
)
|
|
projects.append(project)
|
|
self.save_projects(projects)
|
|
return project
|
|
|
|
def update_project(self, project_id: str, new_name: str = None, new_icon: str = None):
|
|
"""Update a project's name and/or icon."""
|
|
projects = self.load_projects()
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
if new_name is not None:
|
|
p.name = new_name
|
|
if new_icon is not None:
|
|
p.icon = new_icon
|
|
break
|
|
self.save_projects(projects)
|
|
|
|
def delete_project(self, project_id: str,
|
|
callback: Optional[Callable[[], None]] = None):
|
|
"""Delete a project and all its requests (async for secret cleanup)."""
|
|
projects = self.load_projects()
|
|
secret_manager = get_secret_manager()
|
|
|
|
# Remove project from list and save immediately
|
|
projects = [p for p in projects if p.id != project_id]
|
|
self.save_projects(projects)
|
|
|
|
# Delete all secrets for this project asynchronously
|
|
def on_secrets_deleted(success):
|
|
if callback:
|
|
callback()
|
|
|
|
secret_manager.delete_all_project_secrets(project_id, on_secrets_deleted)
|
|
|
|
def add_request(self, project_id: str, name: str, request: HttpRequest, scripts=None) -> SavedRequest:
|
|
"""Add request to a project."""
|
|
projects = self.load_projects()
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
saved_request = SavedRequest(
|
|
id=str(uuid.uuid4()),
|
|
name=name,
|
|
request=request,
|
|
created_at=now,
|
|
modified_at=now,
|
|
scripts=scripts
|
|
)
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
p.requests.append(saved_request)
|
|
break
|
|
self.save_projects(projects)
|
|
return saved_request
|
|
|
|
def find_request_by_name(self, project_id: str, name: str) -> Optional[SavedRequest]:
|
|
"""Find a request by name within a project. Returns None if not found."""
|
|
projects = self.load_projects()
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
for req in p.requests:
|
|
if req.name == name:
|
|
return req
|
|
break
|
|
return None
|
|
|
|
def update_request(self, project_id: str, request_id: str, name: str, request: HttpRequest, scripts=None) -> SavedRequest:
|
|
"""Update an existing request."""
|
|
projects = self.load_projects()
|
|
updated_request = None
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
for req in p.requests:
|
|
if req.id == request_id:
|
|
req.name = name
|
|
req.request = request
|
|
req.scripts = scripts
|
|
req.modified_at = datetime.now(timezone.utc).isoformat()
|
|
updated_request = req
|
|
break
|
|
break
|
|
if updated_request:
|
|
self.save_projects(projects)
|
|
return updated_request
|
|
|
|
def delete_request(self, project_id: str, request_id: str):
|
|
"""Delete a saved request."""
|
|
projects = self.load_projects()
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
p.requests = [r for r in p.requests if r.id != request_id]
|
|
break
|
|
self.save_projects(projects)
|
|
|
|
def add_environment(self, project_id: str, name: str) -> Environment:
|
|
"""Add environment to a project."""
|
|
projects = self.load_projects()
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
environment = None
|
|
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
# Create environment with empty values for all variables
|
|
variables = {var_name: "" for var_name in p.variable_names}
|
|
environment = Environment(
|
|
id=str(uuid.uuid4()),
|
|
name=name,
|
|
variables=variables,
|
|
created_at=now
|
|
)
|
|
p.environments.append(environment)
|
|
break
|
|
|
|
self.save_projects(projects)
|
|
return environment
|
|
|
|
def copy_environment(self, project_id: str, source_env_id: str, new_name: str,
|
|
callback: Optional[Callable] = None):
|
|
"""Copy an environment including all variables (sensitive ones via keyring)."""
|
|
projects = self.load_projects()
|
|
secret_manager = get_secret_manager()
|
|
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
source_env = next((e for e in p.environments if e.id == source_env_id), None)
|
|
if source_env is None:
|
|
if callback:
|
|
callback(None)
|
|
return
|
|
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
new_env_id = str(uuid.uuid4())
|
|
new_env = Environment(
|
|
id=new_env_id,
|
|
name=new_name,
|
|
variables=dict(source_env.variables),
|
|
created_at=now
|
|
)
|
|
p.environments.append(new_env)
|
|
self.save_projects(projects)
|
|
|
|
sensitive_vars = [v for v in p.sensitive_variables if v in source_env.variables]
|
|
if not sensitive_vars:
|
|
if callback:
|
|
callback(new_env)
|
|
return
|
|
|
|
pending_count = [len(sensitive_vars)]
|
|
|
|
def make_copy_handler(var_name, project_name, env_name):
|
|
def on_store_done(success=True):
|
|
pending_count[0] -= 1
|
|
if pending_count[0] == 0 and callback:
|
|
callback(new_env)
|
|
|
|
def on_retrieve(value):
|
|
if value:
|
|
secret_manager.store_secret(
|
|
project_id=project_id,
|
|
environment_id=new_env_id,
|
|
variable_name=var_name,
|
|
value=value,
|
|
project_name=project_name,
|
|
environment_name=env_name,
|
|
callback=lambda success: on_store_done(success)
|
|
)
|
|
else:
|
|
on_store_done()
|
|
|
|
return on_retrieve
|
|
|
|
for var_name in sensitive_vars:
|
|
secret_manager.retrieve_secret(
|
|
project_id=project_id,
|
|
environment_id=source_env_id,
|
|
variable_name=var_name,
|
|
callback=make_copy_handler(var_name, p.name, new_name)
|
|
)
|
|
return
|
|
|
|
if callback:
|
|
callback(None)
|
|
|
|
def update_environment(self, project_id: str, env_id: str, name: str = None, variables: dict = None):
|
|
"""Update an environment's name and/or variables."""
|
|
projects = self.load_projects()
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
for env in p.environments:
|
|
if env.id == env_id:
|
|
if name is not None:
|
|
env.name = name
|
|
if variables is not None:
|
|
env.variables = variables
|
|
break
|
|
break
|
|
self.save_projects(projects)
|
|
|
|
def delete_environment(self, project_id: str, env_id: str,
|
|
callback: Optional[Callable[[], None]] = None):
|
|
"""Delete an environment (async for secret cleanup)."""
|
|
projects = self.load_projects()
|
|
secret_manager = get_secret_manager()
|
|
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
# Remove environment from list
|
|
p.environments = [e for e in p.environments if e.id != env_id]
|
|
break
|
|
|
|
self.save_projects(projects)
|
|
|
|
# Delete all secrets for this environment asynchronously
|
|
def on_secrets_deleted(success):
|
|
if callback:
|
|
callback()
|
|
|
|
secret_manager.delete_all_environment_secrets(project_id, env_id, on_secrets_deleted)
|
|
|
|
def add_variable(self, project_id: str, variable_name: str):
|
|
"""Add a variable to the project and all environments."""
|
|
projects = self.load_projects()
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
if variable_name not in p.variable_names:
|
|
p.variable_names.append(variable_name)
|
|
# Add empty value to all environments
|
|
for env in p.environments:
|
|
env.variables[variable_name] = ""
|
|
break
|
|
self.save_projects(projects)
|
|
|
|
def rename_variable(self, project_id: str, old_name: str, new_name: str,
|
|
callback: Optional[Callable[[], None]] = None):
|
|
"""Rename a variable in the project and all environments (async for secrets)."""
|
|
projects = self.load_projects()
|
|
secret_manager = get_secret_manager()
|
|
is_sensitive = False
|
|
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
# Update variable_names list
|
|
if old_name in p.variable_names:
|
|
idx = p.variable_names.index(old_name)
|
|
p.variable_names[idx] = new_name
|
|
|
|
# Check if sensitive and update list
|
|
if old_name in p.sensitive_variables:
|
|
is_sensitive = True
|
|
idx_sensitive = p.sensitive_variables.index(old_name)
|
|
p.sensitive_variables[idx_sensitive] = new_name
|
|
|
|
# Update all environments
|
|
for env in p.environments:
|
|
if old_name in env.variables:
|
|
env.variables[new_name] = env.variables.pop(old_name)
|
|
break
|
|
|
|
self.save_projects(projects)
|
|
|
|
# Rename secrets in keyring if sensitive
|
|
if is_sensitive:
|
|
secret_manager.rename_variable_secrets(project_id, old_name, new_name, lambda success: callback() if callback else None)
|
|
elif callback:
|
|
callback()
|
|
|
|
def delete_variable(self, project_id: str, variable_name: str,
|
|
callback: Optional[Callable[[], None]] = None):
|
|
"""Delete a variable from the project and all environments (async for secrets)."""
|
|
projects = self.load_projects()
|
|
secret_manager = get_secret_manager()
|
|
is_sensitive = False
|
|
environments_to_cleanup = []
|
|
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
# Remove from variable_names
|
|
if variable_name in p.variable_names:
|
|
p.variable_names.remove(variable_name)
|
|
|
|
# Check if sensitive and get environments for cleanup
|
|
if variable_name in p.sensitive_variables:
|
|
is_sensitive = True
|
|
p.sensitive_variables.remove(variable_name)
|
|
environments_to_cleanup = [env.id for env in p.environments]
|
|
|
|
# Remove from all environments
|
|
for env in p.environments:
|
|
env.variables.pop(variable_name, None)
|
|
break
|
|
|
|
self.save_projects(projects)
|
|
|
|
# Delete secrets for this variable asynchronously
|
|
if is_sensitive and environments_to_cleanup:
|
|
pending_count = [len(environments_to_cleanup)]
|
|
|
|
def on_single_delete(success):
|
|
pending_count[0] -= 1
|
|
if pending_count[0] == 0 and callback:
|
|
callback()
|
|
|
|
for env_id in environments_to_cleanup:
|
|
secret_manager.delete_secret(project_id, env_id, variable_name, on_single_delete)
|
|
elif callback:
|
|
callback()
|
|
|
|
def update_environment_variable(self, project_id: str, env_id: str, variable_name: str, value: str):
|
|
"""Update a single variable value in an environment."""
|
|
projects = self.load_projects()
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
for env in p.environments:
|
|
if env.id == env_id:
|
|
env.variables[variable_name] = value
|
|
break
|
|
break
|
|
self.save_projects(projects)
|
|
|
|
def batch_update_environment_variables(self, project_id: str, env_id: str, variables: dict):
|
|
"""
|
|
Update multiple variables in an environment in one operation.
|
|
|
|
Args:
|
|
project_id: Project ID
|
|
env_id: Environment ID
|
|
variables: Dict of variable_name -> value pairs to update
|
|
"""
|
|
projects = self.load_projects()
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
for env in p.environments:
|
|
if env.id == env_id:
|
|
# Update all variables
|
|
for var_name, var_value in variables.items():
|
|
env.variables[var_name] = var_value
|
|
break
|
|
break
|
|
self.save_projects(projects)
|
|
|
|
# ========== Sensitive Variable Management ==========
|
|
|
|
def mark_variable_as_sensitive(self, project_id: str, variable_name: str,
|
|
callback: Optional[Callable[[bool], None]] = None):
|
|
"""
|
|
Mark a variable as sensitive (move values from JSON to keyring) - async.
|
|
|
|
This will:
|
|
1. Add variable to sensitive_variables list
|
|
2. Move all environment values to keyring
|
|
3. Clear values from JSON (store empty string as placeholder)
|
|
|
|
Args:
|
|
project_id: Project ID
|
|
variable_name: Variable name to mark as sensitive
|
|
callback: Optional callback called with success boolean
|
|
"""
|
|
projects = self.load_projects()
|
|
secret_manager = get_secret_manager()
|
|
secrets_to_store = []
|
|
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
# Add to sensitive list if not already there
|
|
if variable_name not in p.sensitive_variables:
|
|
p.sensitive_variables.append(variable_name)
|
|
|
|
# Collect all environment values to store in keyring
|
|
for env in p.environments:
|
|
if variable_name in env.variables:
|
|
value = env.variables[variable_name]
|
|
secrets_to_store.append({
|
|
'environment_id': env.id,
|
|
'environment_name': env.name,
|
|
'value': value,
|
|
'project_name': p.name
|
|
})
|
|
# Clear from JSON (keep empty placeholder)
|
|
env.variables[variable_name] = ""
|
|
|
|
self.save_projects(projects)
|
|
|
|
# Store all secrets asynchronously
|
|
if not secrets_to_store:
|
|
if callback:
|
|
callback(True)
|
|
return
|
|
|
|
pending_count = [len(secrets_to_store)]
|
|
all_success = [True]
|
|
|
|
def on_single_store(success):
|
|
if not success:
|
|
all_success[0] = False
|
|
pending_count[0] -= 1
|
|
if pending_count[0] == 0 and callback:
|
|
callback(all_success[0])
|
|
|
|
for secret_info in secrets_to_store:
|
|
secret_manager.store_secret(
|
|
project_id=project_id,
|
|
environment_id=secret_info['environment_id'],
|
|
variable_name=variable_name,
|
|
value=secret_info['value'],
|
|
project_name=secret_info['project_name'],
|
|
environment_name=secret_info['environment_name'],
|
|
callback=on_single_store
|
|
)
|
|
return
|
|
|
|
if callback:
|
|
callback(False)
|
|
|
|
def mark_variable_as_nonsensitive(self, project_id: str, variable_name: str,
|
|
callback: Optional[Callable[[bool], None]] = None):
|
|
"""
|
|
Mark a variable as non-sensitive (move values from keyring to JSON) - async.
|
|
|
|
This will:
|
|
1. Remove variable from sensitive_variables list
|
|
2. Move all environment values from keyring to JSON
|
|
3. Delete values from keyring
|
|
|
|
Args:
|
|
project_id: Project ID
|
|
variable_name: Variable name to mark as non-sensitive
|
|
callback: Optional callback called with success boolean
|
|
"""
|
|
projects = self.load_projects()
|
|
secret_manager = get_secret_manager()
|
|
|
|
target_project = None
|
|
environments_to_migrate = []
|
|
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
target_project = p
|
|
# Remove from sensitive list
|
|
if variable_name in p.sensitive_variables:
|
|
p.sensitive_variables.remove(variable_name)
|
|
|
|
environments_to_migrate = list(p.environments)
|
|
break
|
|
|
|
if not target_project or not environments_to_migrate:
|
|
if callback:
|
|
callback(False)
|
|
return
|
|
|
|
# Retrieve all secrets, then update JSON, then delete from keyring
|
|
pending_count = [len(environments_to_migrate)]
|
|
all_success = [True]
|
|
|
|
def on_single_migrate(env):
|
|
def on_retrieve(value):
|
|
# Store in JSON
|
|
env.variables[variable_name] = value or ""
|
|
self.save_projects(projects)
|
|
|
|
# Delete from keyring
|
|
def on_delete(success):
|
|
if not success:
|
|
all_success[0] = False
|
|
pending_count[0] -= 1
|
|
if pending_count[0] == 0 and callback:
|
|
callback(all_success[0])
|
|
|
|
secret_manager.delete_secret(
|
|
project_id=project_id,
|
|
environment_id=env.id,
|
|
variable_name=variable_name,
|
|
callback=on_delete
|
|
)
|
|
return on_retrieve
|
|
|
|
for env in environments_to_migrate:
|
|
secret_manager.retrieve_secret(
|
|
project_id=project_id,
|
|
environment_id=env.id,
|
|
variable_name=variable_name,
|
|
callback=on_single_migrate(env)
|
|
)
|
|
|
|
def is_variable_sensitive(self, project_id: str, variable_name: str) -> bool:
|
|
"""Check if a variable is marked as sensitive."""
|
|
projects = self.load_projects()
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
return variable_name in p.sensitive_variables
|
|
return False
|
|
|
|
def get_variable_value(self, project_id: str, env_id: str, variable_name: str,
|
|
callback: Callable[[str], None]):
|
|
"""
|
|
Get variable value from appropriate storage (keyring or JSON) - async.
|
|
|
|
This is a convenience method that handles both sensitive and non-sensitive variables.
|
|
|
|
Args:
|
|
project_id: Project ID
|
|
env_id: Environment ID
|
|
variable_name: Variable name
|
|
callback: Callback called with variable value (empty string if not found)
|
|
"""
|
|
projects = self.load_projects()
|
|
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
# Check if sensitive
|
|
if variable_name in p.sensitive_variables:
|
|
# Retrieve from keyring asynchronously
|
|
secret_manager = get_secret_manager()
|
|
|
|
def on_retrieve(value):
|
|
callback(value or "")
|
|
|
|
secret_manager.retrieve_secret(
|
|
project_id=project_id,
|
|
environment_id=env_id,
|
|
variable_name=variable_name,
|
|
callback=on_retrieve
|
|
)
|
|
return
|
|
else:
|
|
# Retrieve from JSON synchronously
|
|
for env in p.environments:
|
|
if env.id == env_id:
|
|
callback(env.variables.get(variable_name, ""))
|
|
return
|
|
|
|
callback("")
|
|
|
|
def set_variable_value(self, project_id: str, env_id: str, variable_name: str, value: str,
|
|
callback: Optional[Callable[[bool], None]] = None):
|
|
"""
|
|
Set variable value in appropriate storage (keyring or JSON) - async.
|
|
|
|
This is a convenience method that handles both sensitive and non-sensitive variables.
|
|
|
|
Args:
|
|
project_id: Project ID
|
|
env_id: Environment ID
|
|
variable_name: Variable name
|
|
value: Value to set
|
|
callback: Optional callback called with success boolean
|
|
"""
|
|
projects = self.load_projects()
|
|
secret_manager = get_secret_manager()
|
|
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
# Check if sensitive
|
|
if variable_name in p.sensitive_variables:
|
|
# Store in keyring asynchronously
|
|
for env in p.environments:
|
|
if env.id == env_id:
|
|
secret_manager.store_secret(
|
|
project_id=project_id,
|
|
environment_id=env.id,
|
|
variable_name=variable_name,
|
|
value=value,
|
|
project_name=p.name,
|
|
environment_name=env.name,
|
|
callback=callback
|
|
)
|
|
# Keep empty placeholder in JSON
|
|
env.variables[variable_name] = ""
|
|
self.save_projects(projects)
|
|
return
|
|
else:
|
|
# Store in JSON synchronously
|
|
for env in p.environments:
|
|
if env.id == env_id:
|
|
env.variables[variable_name] = value
|
|
break
|
|
|
|
self.save_projects(projects)
|
|
if callback:
|
|
callback(True)
|
|
return
|
|
|
|
if callback:
|
|
callback(False)
|
|
|
|
def get_environment_with_secrets(self, project_id: str, env_id: str,
|
|
callback: Callable[[Optional[Environment]], None]):
|
|
"""
|
|
Get an environment with all variable values (including secrets from keyring) - async.
|
|
|
|
This is useful for variable substitution - it returns a complete Environment
|
|
object with all values populated from both JSON and keyring.
|
|
|
|
Args:
|
|
project_id: Project ID
|
|
env_id: Environment ID
|
|
callback: Callback called with Environment object (or None if not found)
|
|
"""
|
|
projects = self.load_projects()
|
|
secret_manager = get_secret_manager()
|
|
|
|
for p in projects:
|
|
if p.id == project_id:
|
|
for env in p.environments:
|
|
if env.id == env_id:
|
|
# Create a copy of the environment
|
|
complete_env = Environment(
|
|
id=env.id,
|
|
name=env.name,
|
|
variables=env.variables.copy(),
|
|
created_at=env.created_at
|
|
)
|
|
|
|
# If no sensitive variables, return immediately
|
|
if not p.sensitive_variables:
|
|
callback(complete_env)
|
|
return
|
|
|
|
# Fill in sensitive variable values from keyring
|
|
def on_secrets_retrieved(secrets_dict):
|
|
for var_name, value in secrets_dict.items():
|
|
if value is not None:
|
|
complete_env.variables[var_name] = value
|
|
callback(complete_env)
|
|
|
|
secret_manager.retrieve_multiple_secrets(
|
|
project_id=project_id,
|
|
environment_id=env_id,
|
|
variable_names=list(p.sensitive_variables),
|
|
callback=on_secrets_retrieved
|
|
)
|
|
return
|
|
|
|
callback(None)
|