Source code for kelvin.sdk.services.session
"""SessionService - Manages session state and platform metadata file storage.
This service manages the current session state, including the platform URL,
Docker registry information, and platform metadata by persisting to local files.
"""
from __future__ import annotations
import json
from datetime import datetime
from pathlib import Path
from typing import Final, Optional
import yaml
from platformdirs import user_config_dir
from kelvin.sdk.exceptions import CLIError
from kelvin.sdk.lib.utils.url_utils import normalize_url
from kelvin.sdk.services.session_models import PlatformMetadata, SessionInfo
[docs]
class SessionServiceError(CLIError):
"""Base exception for SessionService errors."""
exit_code: int = 78 # EX_CONFIG - configuration error
[docs]
class SessionNotFoundError(SessionServiceError):
"""Raised when no session is found."""
[docs]
class SessionStorageError(SessionServiceError):
"""Raised when session storage operations fail."""
[docs]
class SessionService:
"""Manages session state and platform metadata file storage.
This is a leaf service with no dependencies. It simply persists session
and metadata information to local files.
Note: Login state checking and token refresh are NOT handled here.
Those responsibilities belong to the command layer (AuthCommands) which
orchestrates CredentialStore and AuthService.
"""
CONFIG_DIR: Final[str] = "kelvin"
SESSION_FILE: Final[str] = "session.yaml"
METADATA_FILE: Final[str] = "metadata.json"
_config_dir: Path
[docs]
def __init__(self) -> None:
"""Initialize SessionService."""
self._config_dir = Path(user_config_dir(self.CONFIG_DIR))
self._cached_session: Optional[SessionInfo] = None
self._cached_metadata: Optional[PlatformMetadata] = None
def _get_session_file_path(self) -> Path:
"""Get the path to the session file."""
return self._config_dir / self.SESSION_FILE
def _get_metadata_file_path(self) -> Path:
"""Get the path to the metadata file."""
return self._config_dir / self.METADATA_FILE
def _ensure_config_dir(self) -> None:
"""Ensure the config directory exists."""
self._config_dir.mkdir(parents=True, exist_ok=True)
[docs]
def get_current_session(self) -> Optional[SessionInfo]:
"""Get the current session information.
Returns:
SessionInfo if a session exists, None otherwise.
"""
if self._cached_session is not None:
return self._cached_session
session_file = self._get_session_file_path()
if not session_file.exists():
return None
try:
with session_file.open("r") as f:
data: object = yaml.safe_load(f) # pyright: ignore[reportAny]
if not isinstance(data, dict):
return None
# Pydantic will validate and coerce types
self._cached_session = SessionInfo.model_validate(data)
return self._cached_session
except (yaml.YAMLError, OSError) as e:
raise SessionStorageError(f"Failed to load session: {e}") from e
except ValueError:
# Invalid data format, return None
return None
[docs]
def set_current_session(self, url: str, metadata: Optional[PlatformMetadata] = None) -> SessionInfo:
"""Set the current session and optionally store metadata.
Args:
url: The platform URL.
metadata: Optional platform metadata. If provided, will also be stored.
Returns:
The created SessionInfo.
Raises:
SessionStorageError: If saving the session fails.
"""
url = normalize_url(url)
session = SessionInfo(
url=url,
docker_url=metadata.docker_url if metadata else None,
docker_port=metadata.docker_port if metadata else None,
docker_path=metadata.docker_path if metadata else None,
last_login=datetime.now(),
)
self._ensure_config_dir()
session_file = self._get_session_file_path()
try:
with session_file.open("w") as f:
yaml.safe_dump(
session.model_dump(mode="json"),
f,
sort_keys=False,
)
except OSError as e:
raise SessionStorageError(f"Failed to save session: {e}") from e
self._cached_session = session
# Also store metadata if provided
if metadata:
self.store_metadata(metadata)
return session
[docs]
def clear_session(self) -> None:
"""Clear the current session and metadata files.
Raises:
SessionStorageError: If clearing the session fails.
"""
session_file = self._get_session_file_path()
metadata_file = self._get_metadata_file_path()
try:
if session_file.exists():
session_file.unlink()
if metadata_file.exists():
metadata_file.unlink()
except OSError as e:
raise SessionStorageError(f"Failed to clear session: {e}") from e
self._cached_session = None
self._cached_metadata = None
[docs]
def get_docker_registry_url(self) -> Optional[str]:
"""Get the Docker registry URL including optional path.
Used for image name prefixes (e.g., ``registry.example.com:5000/kelvin``).
Returns:
Docker registry URL (with optional path) or None if not available.
"""
metadata = self.get_metadata()
if metadata and metadata.docker_url:
url = metadata.docker_url
if metadata.docker_port:
url = f"{url}:{metadata.docker_port}"
if metadata.docker_path:
url = f"{url}/{metadata.docker_path.strip('/')}"
return url
return None
[docs]
def get_docker_login_url(self) -> Optional[str]:
"""Get the Docker registry URL for ``docker login`` (host:port only, no path).
``docker login`` only accepts host:port, not a full image path prefix.
Returns:
Docker registry login URL or None if not available.
"""
metadata = self.get_metadata()
if metadata and metadata.docker_url:
if metadata.docker_port:
return f"{metadata.docker_url}:{metadata.docker_port}"
return metadata.docker_url
return None
[docs]
def get_documentation_url(self) -> Optional[str]:
"""Get the documentation URL.
Returns:
Documentation URL or None if not available.
"""
metadata = self.get_metadata()
return metadata.documentation_url if metadata else None
[docs]
def get_config_dir(self) -> Path:
"""Get the configuration directory path.
Returns:
Path to the configuration directory.
"""
return self._config_dir