Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,9 @@ __pycache__/
*.py[cod]
*$py.class

# Generated version file (created by pdm build)
synodic_client/_version.py

# C extensions
*.so

Expand Down
2 changes: 2 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,8 @@ search_path = ["synodic_client/..."]

[tool.pdm.version]
source = "scm"
write_to = "synodic_client/_version.py"
write_template = "__version__ = '{}'\n"

[tool.pdm.scripts]
analyze = "ruff check synodic_client tests"
Expand Down
7 changes: 7 additions & 0 deletions synodic_client/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,14 @@
UpdateState,
)

# Version is generated at build time by pdm-backend, not committed to repo
try:
from synodic_client._version import __version__
except ImportError:
__version__ = '0.0.0.dev0'

__all__ = [
'__version__',
'Client',
'UpdateChannel',
'UpdateCheckResult',
Expand Down
19 changes: 17 additions & 2 deletions synodic_client/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,12 +28,27 @@ class Client:

@property
def version(self) -> Version:
"""Extracts the version from the installed client
"""Extracts the version from the installed client.

Priority:
1. importlib.metadata
2. _version.py

Returns:
The version data
"""
return Version(importlib.metadata.version(self.distribution))
try:
return Version(importlib.metadata.version(self.distribution))
except importlib.metadata.PackageNotFoundError:
# Frozen executable or missing metadata - use bundled version from SCM
# Import lazily since _version.py is generated at build time and not committed
try:
from synodic_client._version import __version__ as bundled_version # noqa: PLC0415

return Version(bundled_version)
except ImportError:
# Development without build - no version file exists
return Version('0.0.0.dev0')

@property
def package(self) -> str:
Expand Down
177 changes: 75 additions & 102 deletions synodic_client/updater.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,29 @@
"""Self-update functionality using TUF and porringer."""
"""Self-update functionality using TUF and porringer.

This module handles self-updates for synodic-client with two strategies:

1. **Frozen executables** : Uses TUF
for cryptographically verified binary downloads from GitHub releases.
The binary is replaced in-place with automatic backup and rollback support.

2. **Python package installs** : Delegates to porringer for version
checking. Users are instructed to run their package manager's upgrade command
manually, as pip/pipx handle their own security and dependency resolution.
"""

import logging
import shutil
import subprocess
import sys
from collections.abc import Callable
from contextlib import suppress
from dataclasses import dataclass, field
from enum import Enum, auto
from pathlib import Path

from packaging.version import Version
from porringer.api import API
from porringer.schema import CheckUpdateParameters, DownloadParameters, UpdateSource
from porringer.schema import CheckUpdateParameters, UpdateSource
from tuf.api.exceptions import DownloadError, RepositoryError
from tuf.ngclient import Updater as TUFUpdater

Expand Down Expand Up @@ -120,7 +132,7 @@ def check_for_update(self) -> UpdateInfo:
"""Check PyPI for available updates.

Returns:
UpdateInfo with details about available updates
UpdateInfo with details about available updates.
"""
try:
params = CheckUpdateParameters(
Expand All @@ -134,6 +146,7 @@ def check_for_update(self) -> UpdateInfo:

if result.available and result.latest_version:
latest = Version(str(result.latest_version))

self._update_info = UpdateInfo(
available=True,
current_version=self._current_version,
Expand Down Expand Up @@ -164,12 +177,18 @@ def check_for_update(self) -> UpdateInfo:
def download_update(self, progress_callback: Callable | None = None) -> Path | None:
"""Download the update artifact using TUF for verification.

This method is only applicable for frozen executables. For pip/pipx installs,
use the upgrade_command from UpdateInfo instead.

Args:
progress_callback: Optional callback for progress updates (received, total)

Returns:
Path to the downloaded file, or None on failure
"""
if not self.is_frozen:
raise NotImplementedError('Updates for pip/pipx installs are not yet supported')

if self._state != UpdateState.UPDATE_AVAILABLE or not self._update_info:
logger.error('No update available to download')
return None
Expand All @@ -196,9 +215,8 @@ def download_update(self, progress_callback: Callable | None = None) -> Path | N
logger.info('Downloaded and verified update via TUF: %s', download_path)

else:
# Fallback: Direct download via porringer (development mode)
logger.warning('TUF repository not available, using direct download')
self._download_direct(download_path, progress_callback)
# No TUF available - cannot proceed safely for frozen builds
raise RepositoryError('TUF repository not available. Cannot securely download update.')

self._downloaded_path = download_path
self._state = UpdateState.DOWNLOADED
Expand All @@ -218,23 +236,23 @@ def download_update(self, progress_callback: Callable | None = None) -> Path | N
def apply_update(self) -> bool:
"""Apply the downloaded update.

For frozen executables: Replaces the executable with the new version.
For dev mode: Rebuilds with PyInstaller.
This method is only applicable for frozen executables. For pip/pipx installs,
users should run the upgrade_command from UpdateInfo manually.

Returns:
True if update was applied successfully
"""
if not self.is_frozen:
raise NotImplementedError('Updates for pip/pipx installs are not yet supported')

if self._state != UpdateState.DOWNLOADED or not self._downloaded_path:
logger.error('No downloaded update to apply')
return False

self._state = UpdateState.APPLYING

try:
if self.is_frozen:
return self._apply_frozen_update()
else:
return self._apply_dev_update()
return self._apply_frozen_update()

except Exception as e:
logger.exception('Failed to apply update')
Expand Down Expand Up @@ -379,23 +397,6 @@ def _get_backup_path(self) -> Path:
exe_name = self.executable_path.name
return self._config.backup_dir / f'{exe_name}.backup'

def _download_direct(self, download_path: Path, progress_callback: Callable | None = None) -> None:
"""Download update directly via porringer (fallback for dev mode).

Args:
download_path: Destination path
progress_callback: Progress callback
"""
if not self._update_info or not self._update_info.download_url:
raise ValueError('No download URL available')

params = DownloadParameters(
url=self._update_info.download_url,
destination=download_path,
)

self._porringer.update.download(params, progress_callback)

def _apply_frozen_update(self) -> bool:
"""Apply update to a frozen executable.

Expand Down Expand Up @@ -426,91 +427,63 @@ def _apply_frozen_update(self) -> bool:
return True

def _apply_windows_update(self, current_exe: Path, new_exe: Path, backup_path: Path) -> bool:
"""Apply update on Windows using a batch script.
"""Apply update on Windows using rename-then-replace.

Windows allows renaming a running executable but not overwriting it.
We rename the current exe, copy the new one to the original path,
then the app can restart normally. The old exe is cleaned up on next launch.

Args:
current_exe: Path to current executable
new_exe: Path to new executable
backup_path: Path to backup

Returns:
True if update script was created successfully
"""
# Create a batch script that will run after we exit
script_path = self._config.download_dir / 'update.bat'

script_content = f'''@echo off
echo Waiting for application to close...
timeout /t 2 /nobreak > nul
echo Applying update...
copy /y "{new_exe}" "{current_exe}"
if errorlevel 1 (
echo Update failed, restoring backup...
copy /y "{backup_path}" "{current_exe}"
exit /b 1
)
echo Update complete, starting application...
start "" "{current_exe}"
del "%~f0"
'''

script_path.write_text(script_content)

# Schedule the script to run
# Windows-specific process creation flags
flags = 0
if sys.platform == 'win32':
# CREATE_NEW_CONSOLE = 0x00000200, DETACHED_PROCESS = 0x00000008
flags = 0x00000200 | 0x00000008

subprocess.Popen(
['cmd', '/c', str(script_path)],
creationflags=flags,
)

self._state = UpdateState.APPLIED
logger.info('Windows update script scheduled')
return True

def _apply_dev_update(self) -> bool:
"""Apply update in development mode by rebuilding with PyInstaller.
backup_path: Path to backup (already created by caller)

Returns:
True if successful
True if update was applied successfully
"""
logger.info('Development mode: Rebuilding with PyInstaller')

# Find the spec file
spec_file = Path(__file__).parent.parent.parent / 'tool' / 'pyinstaller' / 'synodic.spec'
# Mark the old exe for cleanup (rename it so we can place new one)
old_exe_path = current_exe.with_suffix('.exe.old')

if not spec_file.exists():
raise FileNotFoundError(f'PyInstaller spec file not found: {spec_file}')
# Remove any previous .old file from earlier updates
# May fail if still locked from a very recent restart, that's ok
with suppress(OSError):
if old_exe_path.exists():
old_exe_path.unlink()

# First, update the package
logger.info('Updating package via pip...')
pip_cmd = [sys.executable, '-m', 'pip', 'install', '--upgrade']

if self._config.include_prereleases:
pip_cmd.append('--pre')

pip_cmd.append(self._config.package_name)
try:
# Rename running exe (Windows allows this)
current_exe.rename(old_exe_path)
logger.info('Renamed running executable: %s -> %s', current_exe, old_exe_path)

pip_result = subprocess.run(pip_cmd, capture_output=True, text=True, check=False)
# Copy new exe to original location
shutil.copy2(new_exe, current_exe)
logger.info('Installed new executable: %s', current_exe)

if pip_result.returncode != 0:
raise RuntimeError(f'Pip upgrade failed: {pip_result.stderr}')
self._state = UpdateState.APPLIED
logger.info('Windows update applied successfully (restart required)')
return True

# Rebuild with PyInstaller
logger.info('Rebuilding with PyInstaller...')
pyinstaller_cmd = [sys.executable, '-m', 'PyInstaller', '--clean', str(spec_file)]
except OSError as e:
logger.exception('Failed to apply Windows update via rename')
# Try to restore if rename succeeded but copy failed
if old_exe_path.exists() and not current_exe.exists():
with suppress(OSError):
old_exe_path.rename(current_exe)
raise RuntimeError(f'Windows update failed: {e}') from e

build_result = subprocess.run(
pyinstaller_cmd, capture_output=True, text=True, cwd=spec_file.parent.parent.parent, check=False
)
def cleanup_old_executable(self) -> None:
"""Clean up old executable from previous update.

if build_result.returncode != 0:
raise RuntimeError(f'PyInstaller build failed: {build_result.stderr}')
Call this on application startup to remove the .old file left
from the rename-then-replace update strategy on Windows.
"""
if sys.platform != 'win32' or not self.is_frozen:
return

self._state = UpdateState.APPLIED
logger.info('Development build complete')
return True
old_exe_path = self.executable_path.with_suffix('.exe.old')
if old_exe_path.exists():
try:
old_exe_path.unlink()
logger.info('Cleaned up old executable: %s', old_exe_path)
except OSError as e:
logger.warning('Failed to clean up old executable: %s', e)
Loading
Loading