summaryrefslogtreecommitdiff
path: root/lsfg_vk/base_service.py
diff options
context:
space:
mode:
Diffstat (limited to 'lsfg_vk/base_service.py')
-rw-r--r--lsfg_vk/base_service.py103
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