344 lines
14 KiB
Python
344 lines
14 KiB
Python
# LauncherTool/core/config_manager.py
|
|
"""
|
|
Manages loading and saving application configuration from/to a JSON file.
|
|
"""
|
|
import json
|
|
import os
|
|
import logging
|
|
from .exceptions import (
|
|
ConfigFileNotFoundError,
|
|
ConfigDecodeError,
|
|
ConfigLoadError,
|
|
ConfigSaveError,
|
|
DuplicateNameError,
|
|
NameNotFoundError,
|
|
ApplicationNotFoundError, # Anche se NameNotFoundError potrebbe bastare, essere specifici è ok
|
|
SequenceNotFoundError # Idem
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class ConfigManager:
|
|
"""
|
|
Manages loading and saving configuration from/to a JSON file.
|
|
"""
|
|
|
|
def __init__(self, config_file_path="config.json"):
|
|
"""
|
|
Initializes the ConfigManager with the path to the configuration file.
|
|
|
|
Args:
|
|
config_file_path (str): The absolute path to the configuration JSON file.
|
|
"""
|
|
self.config_file_path = config_file_path
|
|
self.config_data = {} # Dictionary to store configuration data.
|
|
logger.info(f"ConfigManager initialized with file: {self.config_file_path}")
|
|
|
|
def load_config(self):
|
|
"""
|
|
Loads the configuration from the JSON file.
|
|
If the file does not exist, creates a basic empty configuration.
|
|
|
|
Raises:
|
|
ConfigFileNotFoundError: If the file does not exist (and cannot be created for some reason).
|
|
ConfigDecodeError: If the JSON file is malformed.
|
|
ConfigLoadError: For other unexpected errors during loading.
|
|
"""
|
|
try:
|
|
with open(self.config_file_path, 'r', encoding='utf-8') as f:
|
|
self.config_data = json.load(f)
|
|
logger.info(f"Configuration loaded successfully from {self.config_file_path}")
|
|
except FileNotFoundError:
|
|
logger.warning(
|
|
f"Configuration file {self.config_file_path} not found. "
|
|
"Creating a default empty configuration."
|
|
)
|
|
self.config_data = {"applications": [], "sequences": []}
|
|
try:
|
|
self.save_config() # Save the basic configuration immediately.
|
|
logger.info(f"Default configuration saved to {self.config_file_path}")
|
|
except ConfigSaveError as e: # Catch specific save error
|
|
# If saving the default config fails, this is a more serious issue.
|
|
logger.error(f"Failed to save default configuration: {e}")
|
|
raise # Re-raise the ConfigSaveError
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"Error decoding JSON from {self.config_file_path}: {e}")
|
|
raise ConfigDecodeError(self.config_file_path, e) from e
|
|
except Exception as e:
|
|
logger.error(f"An unexpected error occurred while loading config: {e}")
|
|
raise ConfigLoadError(self.config_file_path, e) from e
|
|
|
|
def save_config(self):
|
|
"""
|
|
Saves the current configuration to the JSON file.
|
|
|
|
Raises:
|
|
ConfigSaveError: If an error occurs during saving.
|
|
"""
|
|
try:
|
|
with open(self.config_file_path, 'w', encoding='utf-8') as f:
|
|
json.dump(self.config_data, f, indent=2) # indent for better readability
|
|
logger.info(f"Configuration saved successfully to {self.config_file_path}")
|
|
except Exception as e:
|
|
logger.error(f"Error saving configuration to {self.config_file_path}: {e}")
|
|
raise ConfigSaveError(self.config_file_path, e) from e
|
|
|
|
def get_applications(self):
|
|
"""
|
|
Returns the list of configured applications.
|
|
|
|
Returns:
|
|
list: A list of dictionaries, where each dictionary represents an application.
|
|
Returns an empty list if 'applications' key is not present.
|
|
"""
|
|
return self.config_data.get("applications", [])
|
|
|
|
def get_application_by_name(self, application_name: str):
|
|
"""
|
|
Retrieves a specific application by its name.
|
|
|
|
Args:
|
|
application_name (str): The name of the application to retrieve.
|
|
|
|
Returns:
|
|
dict: The application data.
|
|
|
|
Raises:
|
|
ApplicationNotFoundError: If no application with the given name is found.
|
|
"""
|
|
for app in self.get_applications():
|
|
if app["name"] == application_name:
|
|
return app
|
|
raise ApplicationNotFoundError(application_name)
|
|
|
|
|
|
def get_sequences(self):
|
|
"""
|
|
Returns the list of configured sequences.
|
|
|
|
Returns:
|
|
list: A list of dictionaries, where each dictionary represents a sequence.
|
|
Returns an empty list if 'sequences' key is not present.
|
|
"""
|
|
return self.config_data.get("sequences", [])
|
|
|
|
def get_sequence_by_name(self, sequence_name: str):
|
|
"""
|
|
Retrieves a specific sequence by its name.
|
|
|
|
Args:
|
|
sequence_name (str): The name of the sequence to retrieve.
|
|
|
|
Returns:
|
|
dict: The sequence data.
|
|
|
|
Raises:
|
|
SequenceNotFoundError: If no sequence with the given name is found.
|
|
"""
|
|
for seq in self.get_sequences():
|
|
if seq["name"] == sequence_name:
|
|
return seq
|
|
raise SequenceNotFoundError(sequence_name)
|
|
|
|
def add_application(self, application_data: dict):
|
|
"""
|
|
Adds a new application to the configuration.
|
|
|
|
Args:
|
|
application_data (dict): A dictionary containing the application data.
|
|
Must include a 'name' key.
|
|
|
|
Raises:
|
|
DuplicateNameError: If an application with the same name already exists.
|
|
ConfigSaveError: If saving the configuration fails.
|
|
"""
|
|
applications = self.get_applications()
|
|
if any(app["name"] == application_data["name"] for app in applications):
|
|
logger.warning(
|
|
f"Attempted to add application with duplicate name: {application_data['name']}"
|
|
)
|
|
raise DuplicateNameError("application", application_data["name"])
|
|
|
|
# Ensure 'parameters' key exists, even if empty, for consistency
|
|
if "parameters" not in application_data:
|
|
application_data["parameters"] = []
|
|
|
|
applications.append(application_data)
|
|
self.config_data["applications"] = applications
|
|
self.save_config() # Can raise ConfigSaveError
|
|
logger.info(f"Application '{application_data['name']}' added.")
|
|
|
|
def update_application(self, old_application_name: str, new_application_data: dict):
|
|
"""
|
|
Updates an existing application's data.
|
|
|
|
Args:
|
|
old_application_name (str): The current name of the application to update.
|
|
new_application_data (dict): A dictionary containing the new application data.
|
|
Must include a 'name' key.
|
|
|
|
Raises:
|
|
NameNotFoundError: If the application with 'old_application_name' is not found.
|
|
DuplicateNameError: If 'new_application_data['name']' conflicts with another
|
|
existing application (and it's not the same app being renamed).
|
|
ConfigSaveError: If saving the configuration fails.
|
|
"""
|
|
applications = self.get_applications()
|
|
target_index = -1
|
|
for i, app in enumerate(applications):
|
|
if app["name"] == old_application_name:
|
|
target_index = i
|
|
break
|
|
|
|
if target_index == -1:
|
|
logger.warning(f"Application to update not found: {old_application_name}")
|
|
raise NameNotFoundError("application", old_application_name)
|
|
|
|
# Check for name collision if the name is being changed
|
|
new_name = new_application_data["name"]
|
|
if old_application_name != new_name:
|
|
if any(app["name"] == new_name for i, app in enumerate(applications) if i != target_index):
|
|
logger.warning(
|
|
f"Attempted to rename application '{old_application_name}' to "
|
|
f"'{new_name}', which already exists."
|
|
)
|
|
raise DuplicateNameError("application", new_name)
|
|
|
|
# Ensure 'parameters' key exists in new data
|
|
if "parameters" not in new_application_data:
|
|
new_application_data["parameters"] = []
|
|
|
|
applications[target_index] = new_application_data
|
|
self.config_data["applications"] = applications
|
|
self.save_config() # Can raise ConfigSaveError
|
|
logger.info(f"Application '{old_application_name}' updated to '{new_application_data['name']}'.")
|
|
|
|
def delete_application(self, application_name: str):
|
|
"""
|
|
Deletes an application from the configuration.
|
|
Also removes this application from any steps in sequences.
|
|
|
|
Args:
|
|
application_name (str): The name of the application to delete.
|
|
|
|
Raises:
|
|
NameNotFoundError: If the application to delete is not found.
|
|
ConfigSaveError: If saving the configuration fails.
|
|
"""
|
|
applications = self.get_applications()
|
|
initial_length = len(applications)
|
|
# Filter out the application to be deleted
|
|
self.config_data["applications"] = [
|
|
app for app in applications if app["name"] != application_name
|
|
]
|
|
|
|
if len(self.config_data["applications"]) == initial_length:
|
|
logger.warning(f"Application to delete not found: {application_name}")
|
|
raise NameNotFoundError("application", application_name)
|
|
|
|
# Remove the application from any sequences
|
|
sequences = self.get_sequences()
|
|
for seq in sequences:
|
|
seq["steps"] = [
|
|
step for step in seq.get("steps", []) if step.get("application") != application_name
|
|
]
|
|
self.config_data["sequences"] = sequences
|
|
|
|
self.save_config() # Can raise ConfigSaveError
|
|
logger.info(f"Application '{application_name}' deleted.")
|
|
|
|
def add_sequence(self, sequence_data: dict):
|
|
"""
|
|
Adds a new sequence to the configuration.
|
|
|
|
Args:
|
|
sequence_data (dict): A dictionary containing the sequence data.
|
|
Must include a 'name' key. 'steps' should be a list.
|
|
|
|
Raises:
|
|
DuplicateNameError: If a sequence with the same name already exists.
|
|
ConfigSaveError: If saving the configuration fails.
|
|
"""
|
|
sequences = self.get_sequences()
|
|
if any(seq["name"] == sequence_data["name"] for seq in sequences):
|
|
logger.warning(
|
|
f"Attempted to add sequence with duplicate name: {sequence_data['name']}"
|
|
)
|
|
raise DuplicateNameError("sequence", sequence_data["name"])
|
|
|
|
# Ensure 'steps' key exists and is a list
|
|
if "steps" not in sequence_data or not isinstance(sequence_data["steps"], list):
|
|
sequence_data["steps"] = []
|
|
|
|
sequences.append(sequence_data)
|
|
self.config_data["sequences"] = sequences
|
|
self.save_config() # Can raise ConfigSaveError
|
|
logger.info(f"Sequence '{sequence_data['name']}' added.")
|
|
|
|
def update_sequence(self, old_sequence_name: str, new_sequence_data: dict):
|
|
"""
|
|
Updates an existing sequence's data.
|
|
|
|
Args:
|
|
old_sequence_name (str): The current name of the sequence to update.
|
|
new_sequence_data (dict): A dictionary containing the new sequence data.
|
|
Must include a 'name' key. 'steps' should be a list.
|
|
|
|
Raises:
|
|
NameNotFoundError: If the sequence with 'old_sequence_name' is not found.
|
|
DuplicateNameError: If 'new_sequence_data['name']' conflicts with another
|
|
existing sequence (and it's not the same sequence being renamed).
|
|
ConfigSaveError: If saving the configuration fails.
|
|
"""
|
|
sequences = self.get_sequences()
|
|
target_index = -1
|
|
for i, seq in enumerate(sequences):
|
|
if seq["name"] == old_sequence_name:
|
|
target_index = i
|
|
break
|
|
|
|
if target_index == -1:
|
|
logger.warning(f"Sequence to update not found: {old_sequence_name}")
|
|
raise NameNotFoundError("sequence", old_sequence_name)
|
|
|
|
new_name = new_sequence_data["name"]
|
|
if old_sequence_name != new_name:
|
|
if any(seq["name"] == new_name for i, seq in enumerate(sequences) if i != target_index):
|
|
logger.warning(
|
|
f"Attempted to rename sequence '{old_sequence_name}' to "
|
|
f"'{new_name}', which already exists."
|
|
)
|
|
raise DuplicateNameError("sequence", new_name)
|
|
|
|
# Ensure 'steps' key exists in new data
|
|
if "steps" not in new_sequence_data or not isinstance(new_sequence_data["steps"], list):
|
|
new_sequence_data["steps"] = []
|
|
|
|
sequences[target_index] = new_sequence_data
|
|
self.config_data["sequences"] = sequences
|
|
self.save_config() # Can raise ConfigSaveError
|
|
logger.info(f"Sequence '{old_sequence_name}' updated to '{new_sequence_data['name']}'.")
|
|
|
|
def delete_sequence(self, sequence_name: str):
|
|
"""
|
|
Deletes a sequence from the configuration.
|
|
|
|
Args:
|
|
sequence_name (str): The name of the sequence to delete.
|
|
|
|
Raises:
|
|
NameNotFoundError: If the sequence to delete is not found.
|
|
ConfigSaveError: If saving the configuration fails.
|
|
"""
|
|
sequences = self.get_sequences()
|
|
initial_length = len(sequences)
|
|
self.config_data["sequences"] = [
|
|
seq for seq in sequences if seq["name"] != sequence_name
|
|
]
|
|
|
|
if len(self.config_data["sequences"]) == initial_length:
|
|
logger.warning(f"Sequence to delete not found: {sequence_name}")
|
|
raise NameNotFoundError("sequence", sequence_name)
|
|
|
|
self.save_config() # Can raise ConfigSaveError
|
|
logger.info(f"Sequence '{sequence_name}' deleted.") |