в ruff включён набор UP: современный синтаксис аннотаций

- typing.Dict/List заменены на dict/list, Optional[X] — на X | None;
  эти же места подсвечивал basedpyright в редакторе
- из smtp-convert-secret-key-to-password.py убрана проверка на Python 2
This commit is contained in:
av
2026-08-22 11:37:51 +03:00
parent bdc319df64
commit b5b43a484e
3 changed files with 55 additions and 58 deletions
+52 -54
View File
@@ -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:
+3
View File
@@ -34,6 +34,9 @@ extend-select = [
"ERA",
"PT",
"C90",
# Современный синтаксис: dict/list вместо typing.Dict/List, `X | None`
# вместо Optional. Те же места подсвечивает basedpyright в редакторе.
"UP",
# Наборы, по которым код уже чист: включены, чтобы так и оставалось.
"RET",
"N",
@@ -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."
)