diff --git a/files/backups/backup-all.py b/files/backups/backup-all.py index 406ff2a..958186c 100644 --- a/files/backups/backup-all.py +++ b/files/backups/backup-all.py @@ -26,7 +26,7 @@ from abc import ABC from dataclasses import dataclass, field from datetime import datetime, timedelta from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any import requests from croniter import croniter @@ -87,9 +87,9 @@ class MaintenanceOptions: class Schedule: """Расписание обслуживающих фаз: фаза -> cron-выражение.""" - cron: Dict[str, str] = field(default_factory=dict) + cron: dict[str, str] = field(default_factory=dict) - def due_phases(self, now: datetime) -> List[str]: + def due_phases(self, now: datetime) -> list[str]: """Фазы, которые нужно выполнить в этот прогон, в порядке PHASE_ORDER.""" phases = list(ALWAYS_PHASES) for phase in SCHEDULED_PHASES: @@ -114,14 +114,14 @@ class Schedule: class Application: path: Path owner: str - backup_script: Optional[Path] - backup_targets: List[Path] + backup_script: Path | None + backup_targets: list[Path] @dataclass class BackupResult: success: bool - error: Optional[str] = None + error: str | None = None @dataclass @@ -129,7 +129,7 @@ class StorageRunResult: name: str success: bool duration: float - phases: List[str] + phases: list[str] def format_duration(seconds: float) -> str: @@ -149,8 +149,8 @@ class Storage(ABC): def run( self, - backup_dirs: List[str], - phases: List[str], + backup_dirs: list[str], + phases: list[str], maintenance: MaintenanceOptions, ) -> BackupResult: """Run the requested phases against this storage.""" @@ -160,7 +160,7 @@ class Storage(ABC): class ResticStorage(Storage): TYPE_NAME = "restic" - def __init__(self, name: str, params: Dict[str, Any]) -> None: + def __init__(self, name: str, params: dict[str, Any]) -> None: self.name = name self.restic_repository = str(params.get("restic_repository", "")) self.restic_password = str(params.get("restic_password", "")) @@ -170,7 +170,7 @@ class ResticStorage(Storage): raise ValueError( f"'env' must be a table for storage backend ResticStorage: '{self.name}'" ) - self.env: Dict[str, str] = {str(k): str(v) for k, v in env_raw.items()} + self.env: dict[str, str] = {str(k): str(v) for k, v in env_raw.items()} if not self.restic_repository or not self.restic_password: raise ValueError( @@ -179,8 +179,8 @@ class ResticStorage(Storage): def run( self, - backup_dirs: List[str], - phases: List[str], + backup_dirs: list[str], + phases: list[str], maintenance: MaintenanceOptions, ) -> BackupResult: try: @@ -191,12 +191,12 @@ class ResticStorage(Storage): def __build_steps( self, - backup_dirs: List[str], - phases: List[str], + backup_dirs: list[str], + phases: list[str], maintenance: MaintenanceOptions, - ) -> List[tuple[str, List[str]]]: + ) -> list[tuple[str, list[str]]]: """Собрать restic-команды для запрошенных фаз в порядке PHASE_ORDER.""" - steps: List[tuple[str, List[str]]] = [] + steps: list[tuple[str, list[str]]] = [] for phase in PHASE_ORDER: if phase not in phases: @@ -261,8 +261,8 @@ class ResticStorage(Storage): def __run_internal( self, - backup_dirs: List[str], - phases: List[str], + backup_dirs: list[str], + phases: list[str], maintenance: MaintenanceOptions, ) -> BackupResult: logger.info("Starting restic run for storage '%s'", self.name) @@ -283,9 +283,7 @@ class ResticStorage(Storage): return BackupResult(success=True) - def __run_step( - self, step: str, cmd: List[str], env: Dict[str, str] - ) -> Optional[str]: + def __run_step(self, step: str, cmd: list[str], env: dict[str, str]) -> str | None: """Run a single restic command. Return None on success or error text.""" result = subprocess.run(cmd, env=env, capture_output=True, text=True) @@ -306,7 +304,7 @@ class Notifier(ABC): class AppriseNotifier(Notifier): TYPE_NAME = "apprise" - def __init__(self, name: str, params: Dict[str, Any]) -> None: + def __init__(self, name: str, params: dict[str, Any]) -> None: self.name = name self.api_url = str(params.get("api_url", "")).rstrip("/") self.tag = str(params.get("tag", "")) @@ -336,13 +334,13 @@ class AppriseNotifier(Notifier): class ApplicationFinder: - def __init__(self, roots: List[Path]) -> None: + def __init__(self, roots: list[Path]) -> None: self.roots = roots - self.warnings: List[str] = [] + self.warnings: list[str] = [] - def find_applications(self) -> List[Application]: + def find_applications(self) -> list[Application]: """Discover all applications with their backup scripts and targets.""" - applications: List[Application] = [] + applications: list[Application] = [] source_dirs = itertools.chain(*(root.iterdir() for root in self.roots)) for app_dir in source_dirs: @@ -368,7 +366,7 @@ class ApplicationFinder: applications.sort(key=lambda app: app.path.name) return applications - def _find_backup_script(self, app_dir: Path) -> Optional[Path]: + def _find_backup_script(self, app_dir: Path) -> Path | None: """Find executable backup script in application directory.""" for name in ("backup.sh", "backup"): script_path = app_dir / name @@ -380,10 +378,10 @@ class ApplicationFinder: ) return None - def _find_backup_targets(self, app_dir: Path) -> List[Path]: + def _find_backup_targets(self, app_dir: Path) -> list[Path]: """Resolve backup target directories for an application.""" targets_file = app_dir / BACKUP_TARGETS_FILE - resolved_targets: List[Path] = [] + resolved_targets: list[Path] = [] if targets_file.exists(): for target_line in self._parse_targets_file(targets_file): @@ -411,9 +409,9 @@ class ApplicationFinder: return resolved_targets - def _parse_targets_file(self, targets_file: Path) -> List[str]: + def _parse_targets_file(self, targets_file: Path) -> list[str]: """Parse backup-targets file, skipping comments and empty lines.""" - targets: List[str] = [] + targets: list[str] = [] try: for raw_line in targets_file.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() @@ -431,26 +429,26 @@ class BackupManager: def __init__( self, config: Config, - storages: List[Storage], - notifiers: List[Notifier], + storages: list[Storage], + notifiers: list[Notifier], schedule: Schedule, maintenance: MaintenanceOptions, - forced_phases: Optional[List[str]] = None, + forced_phases: list[str] | None = None, ) -> None: - self.errors: List[str] = [] - self.warnings: List[str] = [] - self.backed_up_apps: List[str] = [] + self.errors: list[str] = [] + self.warnings: list[str] = [] + self.backed_up_apps: list[str] = [] self.config = config self.storages = storages self.notifiers = notifiers self.schedule = schedule self.maintenance = maintenance self.forced_phases = forced_phases - self.active_phases: List[str] = [] + self.active_phases: list[str] = [] self.archive_duration: float = 0.0 - self.storage_results: List[StorageRunResult] = [] + self.storage_results: list[StorageRunResult] = [] - def run_backup_process(self, applications: List[Application]) -> bool: + def run_backup_process(self, applications: list[Application]) -> bool: """Main backup process""" logger.info("Starting backup process") logger.info("Found %d application directories", len(applications)) @@ -473,7 +471,7 @@ class BackupManager: logger.info("Backup completed successfully") return True - def _resolve_phases(self) -> List[str]: + def _resolve_phases(self) -> list[str]: """Какие фазы выполняем в этот прогон: принудительно из CLI или по расписанию.""" if self.forced_phases is not None: logger.info("Phases (forced): %s", ", ".join(self.forced_phases)) @@ -483,7 +481,7 @@ class BackupManager: logger.info("Phases (scheduled): %s", ", ".join(phases)) return phases - def _run_archive_phase(self, applications: List[Application]) -> None: + def _run_archive_phase(self, applications: list[Application]) -> None: """Прогнать скрипты дампов приложений и собрать список того, что уедет в restic. Фаза нужна только вместе с restic backup: без неё дампы делать некому и незачем. @@ -536,9 +534,9 @@ class BackupManager: self.backed_up_apps.append(app_name) @staticmethod - def _collect_backup_dirs(applications: List[Application]) -> List[str]: + def _collect_backup_dirs(applications: list[Application]) -> list[str]: """Собрать цели бекапа всех приложений, сохраняя порядок и убирая дубли.""" - backup_dirs: List[str] = [] + backup_dirs: list[str] = [] for app in applications: for target in app.backup_targets: target_str = str(target) @@ -547,7 +545,7 @@ class BackupManager: logger.info("Found backup directories: %s", backup_dirs) return backup_dirs - def _run_storages(self, backup_dirs: List[str]) -> bool: + def _run_storages(self, backup_dirs: list[str]) -> bool: """Прогнать активные фазы по всем хранилищам. Хранилища независимы: падение одного не отменяет попытку для остальных. @@ -671,7 +669,7 @@ class BackupManager: logger.error("Failed to send notification: %s", e) -def parse_phases(raw: str) -> List[str]: +def parse_phases(raw: str) -> list[str]: """Разобрать CLI-список фаз, вернуть их в порядке PHASE_ORDER.""" requested = {p.strip() for p in raw.split(",") if p.strip()} unknown = requested - set(PHASE_ORDER) @@ -683,10 +681,10 @@ def parse_phases(raw: str) -> List[str]: return [p for p in PHASE_ORDER if p in requested] -def build_storages(raw_config: Dict[str, Any]) -> List[Storage]: +def build_storages(raw_config: dict[str, Any]) -> list[Storage]: """Собрать хранилища из секции [storage] конфига.""" storage_raw = raw_config.get("storage") or {} - storages: List[Storage] = [] + storages: list[Storage] = [] for name, params in storage_raw.items(): if not isinstance(params, dict): raise ValueError(f"Storage config for {name} must be a table") @@ -697,10 +695,10 @@ def build_storages(raw_config: Dict[str, Any]) -> List[Storage]: return storages -def build_notifiers(raw_config: Dict[str, Any]) -> List[Notifier]: +def build_notifiers(raw_config: dict[str, Any]) -> list[Notifier]: """Собрать нотификаторы из секции [notifier] конфига.""" notifications_raw = raw_config.get("notifier") or {} - notifiers: List[Notifier] = [] + notifiers: list[Notifier] = [] for name, params in notifications_raw.items(): if not isinstance(params, dict): raise ValueError(f"Notificator config for {name} must be a table") @@ -711,7 +709,7 @@ def build_notifiers(raw_config: Dict[str, Any]) -> List[Notifier]: return notifiers -def build_schedule(raw_config: Dict[str, Any]) -> Schedule: +def build_schedule(raw_config: dict[str, Any]) -> Schedule: """Собрать расписание обслуживающих фаз из секции [schedule] конфига.""" schedule_raw = raw_config.get("schedule") or {} if not isinstance(schedule_raw, dict): @@ -725,7 +723,7 @@ def build_schedule(raw_config: Dict[str, Any]) -> Schedule: ) -def build_maintenance(raw_config: Dict[str, Any]) -> MaintenanceOptions: +def build_maintenance(raw_config: dict[str, Any]) -> MaintenanceOptions: """Собрать параметры обслуживания из секции [maintenance] конфига.""" maintenance_raw = raw_config.get("maintenance") or {} if not isinstance(maintenance_raw, dict): @@ -744,7 +742,7 @@ def build_maintenance(raw_config: Dict[str, Any]) -> MaintenanceOptions: def initialize( config_path: Path, - forced_phases: Optional[List[str]] = None, + forced_phases: list[str] | None = None, ) -> tuple[ApplicationFinder, BackupManager]: try: with config_path.open("rb") as config_file: diff --git a/pyproject.toml b/pyproject.toml index 35a9bfc..38c0153 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,6 +34,9 @@ extend-select = [ "ERA", "PT", "C90", + # Современный синтаксис: dict/list вместо typing.Dict/List, `X | None` + # вместо Optional. Те же места подсвечивает basedpyright в редакторе. + "UP", # Наборы, по которым код уже чист: включены, чтобы так и оставалось. "RET", "N", diff --git a/scripts/smtp-convert-secret-key-to-password.py b/scripts/smtp-convert-secret-key-to-password.py index cc3dd58..1a679d1 100644 --- a/scripts/smtp-convert-secret-key-to-password.py +++ b/scripts/smtp-convert-secret-key-to-password.py @@ -27,7 +27,6 @@ import hmac import hashlib import base64 import argparse -import sys # These values are required to calculate the signature. Do not change them. @@ -55,9 +54,6 @@ def calculate_key(secret_access_key: str) -> str: def main() -> None: - if sys.version_info[0] < 3: - raise Exception("Must be using Python 3") - parser = argparse.ArgumentParser( description="Convert a Secret Access Key to an SMTP password." )