summaryrefslogtreecommitdiff
path: root/py_modules/lsfg_vk/base_service.py
diff options
context:
space:
mode:
Diffstat (limited to 'py_modules/lsfg_vk/base_service.py')
-rw-r--r--py_modules/lsfg_vk/base_service.py138
1 files changed, 36 insertions, 102 deletions
diff --git a/py_modules/lsfg_vk/base_service.py b/py_modules/lsfg_vk/base_service.py
index 262e2b0..c4978c6 100644
--- a/py_modules/lsfg_vk/base_service.py
+++ b/py_modules/lsfg_vk/base_service.py
@@ -1,132 +1,66 @@
-"""
-Base service class with common functionality.
-"""
-
-import logging
import os
-import shutil
+import tempfile
from pathlib import Path
-from typing import Any, Optional, TypeVar, Dict
+from typing import Any, Dict, Optional, TypeVar
import decky
-from .constants import LOCAL_LIB, LOCAL_SHARE_BASE, VULKAN_LAYER_DIR, SCRIPT_NAME, CONFIG_DIR, CONFIG_FILENAME
+from .constants import CONFIG_DIR, CONFIG_FILENAME, LOCAL_BIN, LOCAL_LIB, SCRIPT_NAME, VULKAN_LAYER_DIR
-ResponseType = TypeVar('ResponseType', bound=Dict[str, Any])
+ResponseType = TypeVar("ResponseType", bound=Dict[str, Any])
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:
- self.log = decky.logger
- else:
- self.log = logger
-
+ self.log = decky.logger if logger is None else logger
self.user_home = Path.home()
+ self.local_bin_dir = self.user_home / LOCAL_BIN
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
self.lsfg_launch_script_path = self.user_home / SCRIPT_NAME
self.config_dir = self.user_home / CONFIG_DIR
self.config_file_path = self.config_dir / CONFIG_FILENAME
-
+
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.config_dir.mkdir(parents=True, exist_ok=True)
- self.log.info(f"Ensured directories exist: {self.local_lib_dir}, {self.local_share_dir}, {self.config_dir}")
-
+ for directory in (self.local_bin_dir, self.local_lib_dir, self.local_share_dir, self.config_dir):
+ directory.mkdir(parents=True, exist_ok=True)
+
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}")
+ if not path.exists() and not path.is_symlink():
return False
-
+ path.unlink()
+ self.log.info(f"Removed {path}")
+ return True
+
def _write_file(self, path: Path, content: str, mode: int = 0o644) -> None:
- """Write content to a file
-
- Args:
- path: Target file path
- content: Content to write
- mode: File permissions (default: 0o644)
-
- Raises:
- OSError: If write fails
- """
+ path.parent.mkdir(parents=True, exist_ok=True)
+ temporary_path = None
try:
- with open(path, 'w', encoding='utf-8') as f:
- f.write(content)
- f.flush()
- os.fsync(f.fileno())
-
- path.chmod(mode)
- self.log.info(f"Wrote to {path}")
-
- except (OSError, IOError, PermissionError) as e:
- self.log.error(f"Failed to write to {path}: {e}")
+ with tempfile.NamedTemporaryFile(
+ mode="w",
+ encoding="utf-8",
+ dir=path.parent,
+ prefix=f".{path.name}.",
+ delete=False,
+ ) as temporary_file:
+ temporary_path = Path(temporary_file.name)
+ temporary_file.write(content)
+ temporary_file.flush()
+ os.fsync(temporary_file.fileno())
+ temporary_path.chmod(mode)
+ os.replace(temporary_path, path)
+ except Exception:
+ if temporary_path is not None:
+ temporary_path.unlink(missing_ok=True)
raise
def _success_response(self, response_type: type, message: str = "", **kwargs) -> Any:
- """Create a standardized success response
-
- Args:
- response_type: The TypedDict response type to create
- message: Success message
- **kwargs: Additional response fields
-
- Returns:
- Success response dict
- """
- response = {
- "success": True,
- "message": message,
- "error": None
- }
+ response = {"success": True, "message": message, "error": None}
response.update(kwargs)
return response
-
+
def _error_response(self, response_type: type, error: str, message: str = "", **kwargs) -> Any:
- """Create a standardized error response
-
- Args:
- response_type: The TypedDict response type to create
- error: Error description
- message: Optional message
- **kwargs: Additional response fields
-
- Returns:
- Error response dict
- """
- response = {
- "success": False,
- "message": message,
- "error": error
- }
+ response = {"success": False, "message": message, "error": error}
response.update(kwargs)
return response