diff options
Diffstat (limited to 'lsfg_vk/base_service.py')
| -rw-r--r-- | lsfg_vk/base_service.py | 103 |
1 files changed, 103 insertions, 0 deletions
diff --git a/lsfg_vk/base_service.py b/lsfg_vk/base_service.py new file mode 100644 index 0000000..b547759 --- /dev/null +++ b/lsfg_vk/base_service.py @@ -0,0 +1,103 @@ +""" +Base service class with common functionality. +""" + +import os +import shutil +import tempfile +from pathlib import Path +from typing import Any, Optional + +from .constants import LOCAL_LIB, LOCAL_SHARE_BASE, VULKAN_LAYER_DIR, SCRIPT_NAME + + +class BaseService: + """Base service class with common functionality""" + + def __init__(self, logger: Optional[Any] = None): + """Initialize base service + + Args: + logger: Logger instance, defaults to decky.logger if None + """ + if logger is None: + import decky + self.log = decky.logger + else: + self.log = logger + + # Initialize common paths using pathlib + self.user_home = Path.home() + self.local_lib_dir = self.user_home / LOCAL_LIB + self.local_share_dir = self.user_home / VULKAN_LAYER_DIR + self.lsfg_script_path = self.user_home / SCRIPT_NAME + + def _ensure_directories(self) -> None: + """Create necessary directories if they don't exist""" + self.local_lib_dir.mkdir(parents=True, exist_ok=True) + self.local_share_dir.mkdir(parents=True, exist_ok=True) + self.log.info(f"Ensured directories exist: {self.local_lib_dir}, {self.local_share_dir}") + + def _remove_if_exists(self, path: Path) -> bool: + """Remove a file if it exists + + Args: + path: Path to the file to remove + + Returns: + True if file was removed, False if it didn't exist + + Raises: + OSError: If removal fails + """ + if path.exists(): + try: + path.unlink() + self.log.info(f"Removed {path}") + return True + except OSError as e: + self.log.error(f"Failed to remove {path}: {e}") + raise + else: + self.log.info(f"File not found: {path}") + return False + + def _atomic_write(self, path: Path, content: str, mode: int = 0o644) -> None: + """Write content to a file atomically + + Args: + path: Target file path + content: Content to write + mode: File permissions (default: 0o644) + + Raises: + OSError: If write fails + """ + # Create temporary file in the same directory to ensure atomic move + temp_path = None + try: + with tempfile.NamedTemporaryFile( + mode='w', + dir=path.parent, + delete=False, + prefix=f'.{path.name}.', + suffix='.tmp' + ) as temp_file: + temp_file.write(content) + temp_path = Path(temp_file.name) + + # Set permissions before moving + temp_path.chmod(mode) + + # Atomic move + temp_path.replace(path) + self.log.info(f"Atomically wrote to {path}") + + except Exception: + # Clean up temp file if something went wrong + if temp_path and temp_path.exists(): + try: + temp_path.unlink() + except OSError: + pass # Best effort cleanup + raise |
