kevinveenbirkenbach/pkgmgr
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style: modernise typing and clean up lint findings

Browse Source 
Repository-wide mechanical cleanup so `ruff check src tests` has a chance
of passing; no behavioural changes.

- Add `from __future__ import annotations` where PEP 604 unions are used.
  This has to come first: pyproject declares requires-python >= 3.9, where
  `X | None` is not evaluable at runtime unless annotations are stringified.
- Replace typing.List/Dict/Tuple/Set with the builtin generics and
  Optional[X] with X | None, then drop the imports that became unused.
  The four actions/*/__init__.py files needed this by hand because ruff
  leaves unused imports in __init__.py alone (possible re-exports).
- Strip shebangs from 72 importable modules. None of them are executable
  or invoked directly; the entry points are console_scripts and runpy.
- Flatten nested `with` blocks, collapse needless-bool returns, and apply
  the remaining mechanical ruff fixes (PIE810, FLY002, PERF102, FURB192,
  RUF059, I001).
- Pass check=False explicitly to the four subprocess.run() calls that
  inspect returncode themselves. That is the existing default.

Two rewrites are visible to mocks, so their tests move with them:
subprocess.run(stdout=PIPE, stderr=PIPE) became capture_output=True, and
open(path, "r", ...) lost the redundant mode.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-Birkenbach
2026-07-27 16:47:13 +02:00
parent 14b95d5639
commit b4e0594901
187 changed files with 591 additions and 856 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/pkgmgr/__init__.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
"""
Top-level pkgmgr package.

1
src/pkgmgr/actions/archive/workflow.py
View File

@@ -2,6 +2,7 @@
from __future__ import annotations
import contextlib
from dataclasses import dataclass
from pathlib import Path

4
src/pkgmgr/actions/branch/close_branch.py
View File

@@ -1,7 +1,5 @@
from __future__ import annotations
from typing import Optional
from pkgmgr.core.git.commands import (
GitDeleteRemoteBranchError,
checkout,
@@ -17,7 +15,7 @@ from pkgmgr.core.git.queries import get_current_branch, resolve_base_branch
def close_branch(
name: Optional[str],
name: str | None,
base_branch: str = "main",
fallback_base: str = "master",
cwd: str = ".",

4
src/pkgmgr/actions/branch/drop_branch.py
View File

@@ -1,7 +1,5 @@
from __future__ import annotations
from typing import Optional
from pkgmgr.core.git.commands import (
GitDeleteRemoteBranchError,
delete_local_branch,
@@ -12,7 +10,7 @@ from pkgmgr.core.git.queries import get_current_branch, resolve_base_branch
def drop_branch(
name: Optional[str],
name: str | None,
base_branch: str = "main",
fallback_base: str = "master",
cwd: str = ".",

4
src/pkgmgr/actions/branch/open_branch.py
View File

@@ -1,7 +1,5 @@
from __future__ import annotations
from typing import Optional
from pkgmgr.core.git.commands import (
checkout,
create_branch,
@@ -13,7 +11,7 @@ from pkgmgr.core.git.queries import resolve_base_branch
def open_branch(
name: Optional[str],
name: str | None,
base_branch: str = "main",
fallback_base: str = "master",
cwd: str = ".",

8
src/pkgmgr/actions/changelog/__init__.py
View File

@@ -1,13 +1,9 @@
#!/usr/bin/env python3
"""
Helpers to generate changelog information from Git history.
"""
from __future__ import annotations
from typing import Optional
from pkgmgr.core.git.queries import (
GitChangelogQueryError,
get_changelog,
@@ -16,8 +12,8 @@ from pkgmgr.core.git.queries import (
def generate_changelog(
cwd: str,
from_ref: Optional[str] = None,
to_ref: Optional[str] = None,
from_ref: str | None = None,
to_ref: str | None = None,
include_merges: bool = False,
) -> str:
"""

18
src/pkgmgr/actions/code_scanning/__init__.py
View File

@@ -27,14 +27,14 @@ class CodeScanningResult:
repo: str
output_dir: str
alert_count: int
files: List[str] = field(default_factory=list)
files: list[str] = field(default_factory=list)
def _gh(args: List[str]) -> subprocess.CompletedProcess[str]:
return subprocess.run(["gh", *args], capture_output=True, text=True)
def _gh(args: list[str]) -> subprocess.CompletedProcess[str]:
return subprocess.run(["gh", *args], capture_output=True, text=True, check=False)
def _resolve_repo(repo: Optional[str]) -> str:
def _resolve_repo(repo: str | None) -> str:
if repo:
return repo
proc = _gh(["repo", "view", "--json", "nameWithOwner", "-q", ".nameWithOwner"])
@@ -48,7 +48,7 @@ def _resolve_repo(repo: Optional[str]) -> str:
return name
def _fetch_json(endpoint: str, params: Optional[List[str]] = None) -> Any:
def _fetch_json(endpoint: str, params: list[str] | None = None) -> Any:
args = ["api", endpoint, "--paginate"]
for param in params or []:
args += ["-f", param]
@@ -74,7 +74,7 @@ def _alert_row(alert: dict) -> str:
return f"- [{severity}] {rule_id} — {path}:{line} ({state})\n {message}"
def _build_summary(repo: str, generated_at: str, alerts: List[dict]) -> str:
def _build_summary(repo: str, generated_at: str, alerts: list[dict]) -> str:
by_severity: Counter = Counter()
by_state: Counter = Counter()
by_rule: Counter = Counter()
@@ -112,9 +112,9 @@ def _build_summary(repo: str, generated_at: str, alerts: List[dict]) -> str:
def download_code_scanning(
repo: Optional[str] = None,
output_dir: Optional[str] = None,
state: Optional[str] = None,
repo: str | None = None,
output_dir: str | None = None,
state: str | None = None,
) -> CodeScanningResult:
if not shutil.which("gh"):
raise CodeScanningError("the GitHub CLI 'gh' is not installed or not on PATH")

2
src/pkgmgr/actions/config/add.py
View File

@@ -30,7 +30,7 @@ def interactive_add(config, USER_CONFIG_PATH: str):
confirm = input("Add this entry to user config? (y/N): ").strip().lower()
if confirm == "y":
if os.path.exists(USER_CONFIG_PATH):
with open(USER_CONFIG_PATH, "r") as f:
with open(USER_CONFIG_PATH) as f:
user_config = yaml.safe_load(f) or {}
else:
user_config = {"repositories": []}

10
src/pkgmgr/actions/config/init.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
"""
Initialize user configuration by scanning the repositories base directory.
@@ -22,7 +20,7 @@ For each discovered repository, the function:
from __future__ import annotations
import os
from typing import Any, Dict
from typing import Any
from pkgmgr.core.command.alias import generate_alias
from pkgmgr.core.config.save import save_user_config
@@ -30,8 +28,8 @@ from pkgmgr.core.git.queries import get_latest_commit
def config_init(
user_config: Dict[str, Any],
defaults_config: Dict[str, Any],
user_config: dict[str, Any],
defaults_config: dict[str, Any],
bin_dir: str,
user_config_path: str,
) -> None:
@@ -128,7 +126,7 @@ def config_init(
"[WARN] Could not read commit (not a git repo or no commits)."
)
entry: Dict[str, Any] = {
entry: dict[str, Any] = {
"provider": provider,
"account": account,
"repository": repo_name,

18
src/pkgmgr/actions/install/__init__.py
View File

@@ -1,6 +1,4 @@
# src/pkgmgr/actions/install/__init__.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
High-level entry point for repository installation.
@@ -16,7 +14,7 @@ Responsibilities:
from __future__ import annotations
import os
from typing import Any, Dict, List, Optional, Tuple
from typing import Any
from pkgmgr.actions.install.context import RepoContext
from pkgmgr.actions.install.installers.makefile import (
@@ -37,7 +35,7 @@ from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier
from pkgmgr.core.repository.verify import verify_repository
Repository = Dict[str, Any]
Repository = dict[str, Any]
INSTALLERS = [
ArchPkgbuildInstaller(),
@@ -52,12 +50,12 @@ INSTALLERS = [
def _ensure_repo_dir(
repo: Repository,
repositories_base_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
preview: bool,
no_verification: bool,
clone_mode: str,
identifier: str,
) -> Optional[str]:
) -> str | None:
"""
Compute and, if necessary, clone the repository directory.
@@ -127,7 +125,7 @@ def _create_context(
repo_dir: str,
repositories_base_dir: str,
bin_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
no_verification: bool,
preview: bool,
quiet: bool,
@@ -155,10 +153,10 @@ def _create_context(
def install_repos(
selected_repos: List[Repository],
selected_repos: list[Repository],
repositories_base_dir: str,
bin_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
no_verification: bool,
preview: bool,
quiet: bool,
@@ -179,7 +177,7 @@ def install_repos(
overall command never exits non-zero because of per-repository failures.
"""
pipeline = InstallationPipeline(INSTALLERS)
failures: List[Tuple[str, str]] = []
failures: list[tuple[str, str]] = []
for repo in selected_repos:
identifier = get_repo_identifier(repo, all_repos)

21
src/pkgmgr/actions/install/capabilities.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
"""
Capability detection for pkgmgr.
@@ -35,7 +33,7 @@ import glob
import os
from abc import ABC, abstractmethod
from collections.abc import Iterable
from typing import TYPE_CHECKING, Optional
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from pkgmgr.actions.install.context import RepoContext
@@ -46,12 +44,12 @@ if TYPE_CHECKING:
# ---------------------------------------------------------------------------
def _read_text_if_exists(path: str) -> Optional[str]:
def _read_text_if_exists(path: str) -> str | None:
"""Read a file as UTF-8 text, returning None if it does not exist or fails."""
if not os.path.exists(path):
return None
try:
with open(path, "r", encoding="utf-8") as f:
with open(path, encoding="utf-8") as f:
return f.read()
except OSError:
return None
@@ -75,12 +73,12 @@ def _scan_files_for_patterns(files: Iterable[str], patterns: Iterable[str]) -> b
return False
def _first_spec_file(repo_dir: str) -> Optional[str]:
def _first_spec_file(repo_dir: str) -> str | None:
"""Return the first *.spec file in repo_dir, if any."""
matches = glob.glob(os.path.join(repo_dir, "*.spec"))
if not matches:
return None
return sorted(matches)[0]
return min(matches)
# ---------------------------------------------------------------------------
@@ -216,7 +214,7 @@ class MakeInstallCapability(CapabilityMatcher):
if not os.path.exists(makefile):
return False
try:
with open(makefile, "r", encoding="utf-8") as f:
with open(makefile, encoding="utf-8") as f:
for line in f:
if line.strip().startswith("install:"):
return True
@@ -360,7 +358,7 @@ def detect_capabilities(
def resolve_effective_capabilities(
ctx: RepoContext,
layers: Optional[Iterable[str]] = None,
layers: Iterable[str] | None = None,
) -> dict[str, set[str]]:
"""
Resolve *effective* capabilities for each layer using a bottom-up strategy.
@@ -381,10 +379,7 @@ def resolve_effective_capabilities(
This means *any* higher layer can overshadow a lower layer, not just
a specific one like Nix. The resolver is completely generic.
"""
if layers is None:
layers_list = list(LAYER_ORDER)
else:
layers_list = list(layers)
layers_list = list(LAYER_ORDER) if layers is None else list(layers)
raw_caps = detect_capabilities(ctx, layers_list)
effective: dict[str, set[str]] = {layer: set() for layer in layers_list}

8
src/pkgmgr/actions/install/context.py
View File

@@ -1,6 +1,4 @@
# src/pkgmgr/actions/install/context.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Shared context object for repository installation steps.
@@ -10,19 +8,19 @@ they do not depend on global state or long parameter lists.
"""
from dataclasses import dataclass
from typing import Any, Dict, List
from typing import Any
@dataclass
class RepoContext:
"""Container for all repository-related data used during installation."""
repo: Dict[str, Any]
repo: dict[str, Any]
identifier: str
repo_dir: str
repositories_base_dir: str
bin_dir: str
all_repos: List[Dict[str, Any]]
all_repos: list[dict[str, Any]]
no_verification: bool
preview: bool

2
src/pkgmgr/actions/install/installers/__init__.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
"""
Installer package for pkgmgr.

10
src/pkgmgr/actions/install/installers/base.py
View File

@@ -1,11 +1,9 @@
#!/usr/bin/env python3
"""
Base interface for all installer components in the pkgmgr installation pipeline.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Optional, Set
from pkgmgr.actions.install.capabilities import CAPABILITY_MATCHERS
from pkgmgr.actions.install.context import RepoContext
@@ -23,9 +21,9 @@ class BaseInstaller(ABC):
# Examples: "nix", "python", "makefile".
# This is used by capability matchers to decide which patterns to
# search for in the repository.
layer: Optional[str] = None
layer: str | None = None
def discover_capabilities(self, ctx: RepoContext) -> Set[str]:
def discover_capabilities(self, ctx: RepoContext) -> set[str]:
"""
Determine which logical capabilities this installer will provide
for this specific repository instance.
@@ -35,7 +33,7 @@ class BaseInstaller(ABC):
Makefile, etc.) and decide, via string matching, whether a given
capability is actually provided by this layer.
"""
caps: Set[str] = set()
caps: set[str] = set()
if not self.layer:
return caps

6
src/pkgmgr/actions/install/installers/makefile.py
View File

@@ -26,16 +26,14 @@ class MakefileInstaller(BaseInstaller):
def _has_install_target(self, makefile_path: str) -> bool:
try:
with open(makefile_path, "r", encoding="utf-8", errors="ignore") as f:
with open(makefile_path, encoding="utf-8", errors="ignore") as f:
content = f.read()
except OSError:
return False
if re.search(r"^install\s*:", content, flags=re.MULTILINE):
return True
if re.search(r"^install-[a-zA-Z0-9_-]*\s*:", content, flags=re.MULTILINE):
return True
return False
return bool(re.search(r"^install-[a-zA-Z0-9_-]*\s*:", content, flags=re.MULTILINE))
def run(self, ctx: RepoContext) -> None:
makefile_path = os.path.join(ctx.repo_dir, self.MAKEFILE_NAME)

4
src/pkgmgr/actions/install/installers/nix/conflicts.py
View File

@@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING, List
from typing import TYPE_CHECKING
from .profile import NixProfileInspector
from .retry import GitHubRateLimitRetry
@@ -49,7 +49,7 @@ class NixConflictResolver:
store_prefixes = self._parser.existing_store_prefixes(combined)
# 2) Resolve them to concrete remove tokens
tokens: List[str] = self._profile.find_remove_tokens_for_store_prefixes(
tokens: list[str] = self._profile.find_remove_tokens_for_store_prefixes(
ctx,
self._runner,
store_prefixes,

4
src/pkgmgr/actions/install/installers/nix/installer.py
View File

@@ -2,7 +2,7 @@ from __future__ import annotations
import os
import shutil
from typing import TYPE_CHECKING, List, Tuple
from typing import TYPE_CHECKING
from pkgmgr.actions.install.installers.base import BaseInstaller
@@ -42,7 +42,7 @@ class NixFlakeInstaller(BaseInstaller):
return os.path.exists(os.path.join(ctx.repo_dir, self.FLAKE_FILE))
def _profile_outputs(self, ctx: RepoContext) -> List[Tuple[str, bool]]:
def _profile_outputs(self, ctx: RepoContext) -> list[tuple[str, bool]]:
# (output_name, allow_failure)
if ctx.identifier in {"pkgmgr", "package-manager"}:
return [("pkgmgr", False), ("default", True)]

24
src/pkgmgr/actions/install/installers/nix/profile/inspector.py
View File

@@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, List
from typing import TYPE_CHECKING, Any
from .matcher import (
entry_matches_output,
@@ -43,11 +43,11 @@ class NixProfileInspector:
ctx: RepoContext,
runner: CommandRunner,
output: str,
) -> List[int]:
) -> list[int]:
data = self.list_json(ctx, runner)
entries = normalize_elements(data)
hits: List[int] = []
hits: list[int] = []
for e in entries:
if e.index is None:
continue
@@ -61,7 +61,7 @@ class NixProfileInspector:
ctx: RepoContext,
runner: CommandRunner,
store_path: str,
) -> List[int]:
) -> list[int]:
needle = (store_path or "").strip()
if not needle:
return []
@@ -69,7 +69,7 @@ class NixProfileInspector:
data = self.list_json(ctx, runner)
entries = normalize_elements(data)
hits: List[int] = []
hits: list[int] = []
for e in entries:
if e.index is None:
continue
@@ -87,7 +87,7 @@ class NixProfileInspector:
ctx: RepoContext,
runner: CommandRunner,
output: str,
) -> List[str]:
) -> list[str]:
"""
Returns profile remove tokens to remove entries matching a given output.
@@ -101,7 +101,7 @@ class NixProfileInspector:
data = self.list_json(ctx, runner)
entries = normalize_elements(data)
tokens: List[str] = [
tokens: list[str] = [
out
] # critical: matches nix's own suggestion for conflicts
@@ -119,7 +119,7 @@ class NixProfileInspector:
# stable unique preserving order
seen: set[str] = set()
uniq: List[str] = []
uniq: list[str] = []
for t in tokens:
if t and t not in seen:
uniq.append(t)
@@ -130,8 +130,8 @@ class NixProfileInspector:
self,
ctx: RepoContext,
runner: CommandRunner,
prefixes: List[str],
) -> List[str]:
prefixes: list[str],
) -> list[str]:
"""
Returns remove tokens for entries whose store path matches any prefix.
"""
@@ -143,7 +143,7 @@ class NixProfileInspector:
data = self.list_json(ctx, runner)
entries = normalize_elements(data)
tokens: List[str] = []
tokens: list[str] = []
for e in entries:
if not e.store_paths:
continue
@@ -156,7 +156,7 @@ class NixProfileInspector:
tokens.append(n)
seen: set[str] = set()
uniq: List[str] = []
uniq: list[str] = []
for t in tokens:
if t and t not in seen:
uniq.append(t)

6
src/pkgmgr/actions/install/installers/nix/profile/matcher.py
View File

@@ -1,7 +1,5 @@
from __future__ import annotations
from typing import List
from .models import NixProfileEntry
@@ -51,9 +49,9 @@ def entry_matches_store_path(entry: NixProfileEntry, store_path: str) -> bool:
return any((p or "") == needle for p in entry.store_paths)
def stable_unique_ints(values: List[int]) -> List[int]:
def stable_unique_ints(values: list[int]) -> list[int]:
seen: set[int] = set()
uniq: List[int] = []
uniq: list[int] = []
for v in values:
if v in seen:
continue

5
src/pkgmgr/actions/install/installers/nix/profile/models.py
View File

@@ -1,7 +1,6 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import List, Optional
@dataclass(frozen=True)
@@ -11,7 +10,7 @@ class NixProfileEntry:
"""
key: str
index: Optional[int]
index: int | None
name: str
attr_path: str
store_paths: List[str]
store_paths: list[str]

12
src/pkgmgr/actions/install/installers/nix/profile/normalizer.py
View File

@@ -2,12 +2,12 @@ from __future__ import annotations
import re
from collections.abc import Iterable
from typing import Any, Dict, List, Optional
from typing import Any
from .models import NixProfileEntry
def coerce_index(key: str, entry: Dict[str, Any]) -> Optional[int]:
def coerce_index(key: str, entry: dict[str, Any]) -> int | None:
"""
Nix JSON schema varies:
- elements keys might be "0", "1", ...
@@ -73,7 +73,7 @@ def iter_store_paths(entry: Dict[str, Any]) -> Iterable[str]:
outs = entry.get("outputs")
if isinstance(outs, dict):
for _, ov in outs.items():
for ov in outs.values():
if isinstance(ov, dict):
p = ov.get("storePath")
if isinstance(p, str):
@@ -88,7 +88,7 @@ def normalize_store_path(store_path: str) -> str:
return (store_path or "").strip()
def normalize_elements(data: Dict[str, Any]) -> List[NixProfileEntry]:
def normalize_elements(data: dict[str, Any]) -> list[NixProfileEntry]:
"""
Converts nix profile list JSON into a list of normalized entries.
@@ -100,7 +100,7 @@ def normalize_elements(data: Dict[str, Any]) -> List[NixProfileEntry]:
if not isinstance(elements, dict):
return []
normalized: List[NixProfileEntry] = []
normalized: list[NixProfileEntry] = []
for k, entry in elements.items():
if not isinstance(entry, dict):
@@ -110,7 +110,7 @@ def normalize_elements(data: Dict[str, Any]) -> List[NixProfileEntry]:
name = str(entry.get("name", "") or "")
attr = str(entry.get("attrPath", "") or "")
store_paths: List[str] = []
store_paths: list[str] = []
for p in iter_store_paths(entry):
sp = normalize_store_path(p)
if sp:

4
src/pkgmgr/actions/install/installers/nix/profile/parser.py
View File

@@ -1,10 +1,10 @@
from __future__ import annotations
import json
from typing import Any, Dict
from typing import Any
def parse_profile_list_json(raw: str) -> Dict[str, Any]:
def parse_profile_list_json(raw: str) -> dict[str, Any]:
"""
Parse JSON output from `nix profile list --json`.

14
src/pkgmgr/actions/install/installers/nix/profile_list.py
View File

@@ -1,7 +1,7 @@
from __future__ import annotations
import re
from typing import TYPE_CHECKING, List, Tuple
from typing import TYPE_CHECKING
from .runner import CommandRunner
@@ -19,12 +19,12 @@ class NixProfileListReader:
m = re.match(r"^(/nix/store/[0-9a-z]{32}-[^/ \t]+)", raw)
return m.group(1) if m else raw
def entries(self, ctx: RepoContext) -> List[Tuple[int, str]]:
def entries(self, ctx: RepoContext) -> list[tuple[int, str]]:
res = self._runner.run(ctx, "nix profile list", allow_failure=True)
if res.returncode != 0:
return []
entries: List[Tuple[int, str]] = []
entries: list[tuple[int, str]] = []
pat = re.compile(
r"^\s*(\d+)\s+.*?(/nix/store/[0-9a-z]{32}-[^/ \t]+)",
re.MULTILINE,
@@ -49,20 +49,20 @@ class NixProfileListReader:
return uniq
def indices_matching_store_prefixes(
self, ctx: RepoContext, prefixes: List[str]
) -> List[int]:
self, ctx: RepoContext, prefixes: list[str]
) -> list[int]:
prefixes = [self._store_prefix(p) for p in prefixes if p]
prefixes = [p for p in prefixes if p]
if not prefixes:
return []
hits: List[int] = []
hits: list[int] = []
for idx, sp in self.entries(ctx):
if any(sp == p for p in prefixes):
hits.append(idx)
seen: set[int] = set()
uniq: List[int] = []
uniq: list[int] = []
for i in hits:
if i not in seen:
seen.add(i)

3
src/pkgmgr/actions/install/installers/nix/runner.py
View File

@@ -31,8 +31,7 @@ class CommandRunner:
shell=True,
cwd=repo_dir,
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
capture_output=True,
text=True,
)
except Exception as e:

13
src/pkgmgr/actions/install/installers/nix/textparse.py
View File

@@ -1,7 +1,6 @@
from __future__ import annotations
import re
from typing import List
class NixConflictTextParser:
@@ -11,13 +10,13 @@ class NixConflictTextParser:
m = re.match(r"^(/nix/store/[0-9a-z]{32}-[^/ \t]+)", raw)
return m.group(1) if m else raw
def remove_tokens(self, text: str) -> List[str]:
def remove_tokens(self, text: str) -> list[str]:
pat = re.compile(
r"^\s*nix profile remove\s+([^\s'\"`]+|'[^']+'|\"[^\"]+\")\s*$",
re.MULTILINE,
)
tokens: List[str] = []
tokens: list[str] = []
for m in pat.finditer(text or ""):
t = (m.group(1) or "").strip()
if (t.startswith("'") and t.endswith("'")) or (
@@ -28,7 +27,7 @@ class NixConflictTextParser:
tokens.append(t)
seen: set[str] = set()
uniq: List[str] = []
uniq: list[str] = []
for t in tokens:
if t not in seen:
seen.add(t)
@@ -36,9 +35,9 @@ class NixConflictTextParser:
return uniq
def existing_store_prefixes(self, text: str) -> List[str]:
def existing_store_prefixes(self, text: str) -> list[str]:
lines = (text or "").splitlines()
prefixes: List[str] = []
prefixes: list[str] = []
in_existing = False
in_new = False
@@ -69,7 +68,7 @@ class NixConflictTextParser:
norm = [self._store_prefix(p) for p in prefixes if p]
seen: set[str] = set()
uniq: List[str] = []
uniq: list[str] = []
for p in norm:
if p and p not in seen:
seen.add(p)

8
src/pkgmgr/actions/install/installers/os_packages/debian_control.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
"""
Installer for Debian/Ubuntu packages defined via debian/control.
@@ -12,11 +10,11 @@ This installer:
It is intended for Debian-based systems where dpkg-buildpackage and
apt/dpkg tooling are available.
"""
from __future__ import annotations
import glob
import os
import shutil
from typing import List, Optional
from pkgmgr.actions.install.context import RepoContext
from pkgmgr.actions.install.installers.base import BaseInstaller
@@ -55,7 +53,7 @@ class DebianControlInstaller(BaseInstaller):
return os.path.exists(self._control_path(ctx))
def _find_built_debs(self, repo_dir: str) -> List[str]:
def _find_built_debs(self, repo_dir: str) -> list[str]:
"""
Find .deb files built by dpkg-buildpackage.
@@ -66,7 +64,7 @@ class DebianControlInstaller(BaseInstaller):
pattern = os.path.join(parent, "*.deb")
return sorted(glob.glob(pattern))
def _privileged_prefix(self) -> Optional[str]:
def _privileged_prefix(self) -> str | None:
"""
Determine how to run privileged commands:

14
src/pkgmgr/actions/install/installers/os_packages/rpm_spec.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
"""
Installer for RPM-based packages defined in *.spec files.
@@ -13,12 +11,12 @@ This installer:
It targets RPM-based systems (Fedora / RHEL / CentOS / Rocky / Alma, etc.).
"""
from __future__ import annotations
import glob
import os
import shutil
import tarfile
from typing import List, Optional, Tuple
from pkgmgr.actions.install.context import RepoContext
from pkgmgr.actions.install.installers.base import BaseInstaller
@@ -52,7 +50,7 @@ class RpmSpecInstaller(BaseInstaller):
return has_dnf or has_yum or has_yum_builddep
def _spec_path(self, ctx: RepoContext) -> Optional[str]:
def _spec_path(self, ctx: RepoContext) -> str | None:
"""Return the first *.spec file in the repository root, if any."""
pattern = os.path.join(ctx.repo_dir, "*.spec")
matches = sorted(glob.glob(pattern))
@@ -91,7 +89,7 @@ class RpmSpecInstaller(BaseInstaller):
for sub in ("BUILD", "BUILDROOT", "RPMS", "SOURCES", "SPECS", "SRPMS"):
os.makedirs(os.path.join(topdir, sub), exist_ok=True)
def _parse_name_version(self, spec_path: str) -> Optional[Tuple[str, str]]:
def _parse_name_version(self, spec_path: str) -> tuple[str, str] | None:
"""
Parse Name and Version from the given .spec file.
@@ -100,7 +98,7 @@ class RpmSpecInstaller(BaseInstaller):
name = None
version = None
with open(spec_path, "r", encoding="utf-8") as f:
with open(spec_path, encoding="utf-8") as f:
for raw_line in f:
line = raw_line.strip()
# Ignore comments
@@ -182,7 +180,7 @@ class RpmSpecInstaller(BaseInstaller):
return self._spec_path(ctx) is not None
def _find_built_rpms(self) -> List[str]:
def _find_built_rpms(self) -> list[str]:
"""
Find RPMs built by rpmbuild.
@@ -212,7 +210,7 @@ class RpmSpecInstaller(BaseInstaller):
run_command(cmd, cwd=ctx.repo_dir, preview=ctx.preview)
def _install_built_rpms(self, ctx: RepoContext, rpms: List[str]) -> None:
def _install_built_rpms(self, ctx: RepoContext, rpms: list[str]) -> None:
"""
Install or upgrade the built RPMs.

11
src/pkgmgr/actions/install/layers.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
"""
CLI layer model for the pkgmgr installation pipeline.
@@ -18,7 +16,6 @@ from __future__ import annotations
import os
from enum import Enum
from typing import Optional
class CliLayer(str, Enum):
@@ -37,7 +34,7 @@ CLI_LAYERS: list[CliLayer] = [
]
def layer_priority(layer: Optional[CliLayer]) -> int:
def layer_priority(layer: CliLayer | None) -> int:
"""
Return a numeric priority index for a given layer.
@@ -69,13 +66,11 @@ def classify_command_layer(command: str, repo_dir: str) -> CliLayer:
home = os.path.expanduser("~")
# OS package managers
if command_abs.startswith("/usr/") or command_abs.startswith("/bin/"):
if command_abs.startswith(("/usr/", "/bin/")):
return CliLayer.OS_PACKAGES
# Nix store / profile
if command_abs.startswith("/nix/store/") or command_abs.startswith(
os.path.join(home, ".nix-profile")
):
if command_abs.startswith(("/nix/store/", os.path.join(home, ".nix-profile"))):
return CliLayer.NIX
# User-local bin

9
src/pkgmgr/actions/install/pipeline.py
View File

@@ -1,6 +1,4 @@
# src/pkgmgr/actions/install/pipeline.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Installation pipeline orchestration for repositories.
@@ -10,7 +8,6 @@ from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Optional, Set
from pkgmgr.actions.install.context import RepoContext
from pkgmgr.actions.install.installers.base import BaseInstaller
@@ -25,8 +22,8 @@ from pkgmgr.core.command.resolve import resolve_command_for_repo
@dataclass
class CommandState:
command: Optional[str]
layer: Optional[CliLayer]
command: str | None
layer: CliLayer | None
class CommandResolver:
@@ -84,7 +81,7 @@ class InstallationPipeline:
else:
repo.pop("command", None)
provided_capabilities: Set[str] = set()
provided_capabilities: set[str] = set()
for installer in self._installers:
layer_name = getattr(installer, "layer", None)

4
src/pkgmgr/actions/mirror/context.py
View File

@@ -1,7 +1,5 @@
from __future__ import annotations
from typing import List
from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier
@@ -12,7 +10,7 @@ from .types import MirrorMap, RepoMirrorContext, Repository
def build_context(
repo: Repository,
repositories_base_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
) -> RepoMirrorContext:
"""
Build a RepoMirrorContext for a single repository.

6
src/pkgmgr/actions/mirror/diff_cmd.py
View File

@@ -1,16 +1,14 @@
from __future__ import annotations
from typing import List
from .context import build_context
from .printing import print_header
from .types import Repository
def diff_mirrors(
selected_repos: List[Repository],
selected_repos: list[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
) -> None:
"""
Show differences between config mirrors and MIRRORS file.

14
src/pkgmgr/actions/mirror/git_remote.py
View File

@@ -1,7 +1,6 @@
from __future__ import annotations
import os
from typing import Optional, Set
from pkgmgr.core.git.commands import (
GitAddRemoteError,
@@ -38,13 +37,10 @@ def _is_git_remote_url(url: str) -> bool:
if u.startswith("ssh://"):
return True
if (u.startswith("https://") or u.startswith("http://")) and u.endswith(".git"):
return True
return False
return bool((u.startswith(("https://", "http://"))) and u.endswith(".git"))
def build_default_ssh_url(repo: Repository) -> Optional[str]:
def build_default_ssh_url(repo: Repository) -> str | None:
provider = repo.get("provider")
account = repo.get("account")
name = repo.get("repository")
@@ -66,7 +62,7 @@ def _git_mirrors_only(m: MirrorMap) -> MirrorMap:
def determine_primary_remote_url(
repo: Repository,
ctx: RepoMirrorContext,
) -> Optional[str]:
) -> str | None:
"""
Priority order (GIT URLS ONLY):
1. origin from resolved mirrors (if it is a git URL)
@@ -80,7 +76,7 @@ def determine_primary_remote_url(
return origin
for mirrors in (ctx.file_mirrors, ctx.config_mirrors):
for _, url in mirrors.items():
for url in mirrors.values():
if url and _is_git_remote_url(url):
return url
@@ -116,7 +112,7 @@ def _ensure_additional_push_urls(
Non-git URLs (like PyPI) are ignored and will never land in git config.
"""
git_only = _git_mirrors_only(mirrors)
desired: Set[str] = {u for u in git_only.values() if u and u != primary}
desired: set[str] = {u for u in git_only.values() if u and u != primary}
if not desired:
return

2
src/pkgmgr/actions/mirror/io.py
View File

@@ -42,7 +42,7 @@ def read_mirrors_file(repo_dir: str, filename: str = "MIRRORS") -> MirrorMap:
return mirrors
try:
with open(path, "r", encoding="utf-8") as fh:
with open(path, encoding="utf-8") as fh:
for line in fh:
stripped = line.strip()
if not stripped or stripped.startswith("#"):

6
src/pkgmgr/actions/mirror/list_cmd.py
View File

@@ -1,16 +1,14 @@
from __future__ import annotations
from typing import List
from .context import build_context
from .printing import print_header, print_named_mirrors
from .types import Repository
def list_mirrors(
selected_repos: List[Repository],
selected_repos: list[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
source: str = "all",
) -> None:
"""

17
src/pkgmgr/actions/mirror/merge_cmd.py
View File

@@ -1,7 +1,6 @@
from __future__ import annotations
import os
from typing import Dict, List, Optional, Tuple
import yaml
@@ -16,7 +15,7 @@ from .types import MirrorMap, Repository
# -----------------------------------------------------------------------------
def _repo_key(repo: Repository) -> Tuple[str, str, str]:
def _repo_key(repo: Repository) -> tuple[str, str, str]:
"""
Normalised key for identifying a repository in config files.
"""
@@ -27,7 +26,7 @@ def _repo_key(repo: Repository) -> Tuple[str, str, str]:
)
def _load_user_config(path: str) -> Dict[str, object]:
def _load_user_config(path: str) -> dict[str, object]:
"""
Load a user config YAML file as dict.
Non-dicts yield {}.
@@ -49,13 +48,13 @@ def _load_user_config(path: str) -> Dict[str, object]:
def merge_mirrors(
selected_repos: List[Repository],
selected_repos: list[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
source: str,
target: str,
preview: bool = False,
user_config_path: Optional[str] = None,
user_config_path: str | None = None,
) -> None:
"""
Merge mirrors between config and MIRRORS file.
@@ -73,8 +72,8 @@ def merge_mirrors(
"""
# Load user config once if we intend to write to it.
user_cfg: Optional[Dict[str, object]] = None
user_cfg_path_expanded: Optional[str] = None
user_cfg: dict[str, object] | None = None
user_cfg_path_expanded: str | None = None
if target == "config" and user_config_path and not preview:
user_cfg_path_expanded = os.path.expanduser(user_config_path)
@@ -129,7 +128,7 @@ def merge_mirrors(
repos = user_cfg.get("repositories")
target_key = _repo_key(repo)
existing_repo: Optional[Repository] = None
existing_repo: Repository | None = None
# Find existing repo entry
for entry in repos:

4
src/pkgmgr/actions/mirror/remote_provision.py
View File

@@ -1,7 +1,5 @@
from __future__ import annotations
from typing import List
from pkgmgr.core.remote_provisioning import ProviderHint, RepoSpec, ensure_remote_repo
from pkgmgr.core.remote_provisioning.ensure import EnsureOptions
@@ -64,7 +62,7 @@ def ensure_remote_repository_for_url(
def ensure_remote_repository(
repo: Repository,
repositories_base_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
preview: bool,
) -> None:
"""

14
src/pkgmgr/actions/mirror/setup_cmd.py
View File

@@ -1,7 +1,5 @@
from __future__ import annotations
from typing import List
from pkgmgr.core.git.queries import probe_remote_reachable_detail
from pkgmgr.core.remote_provisioning import ProviderHint, RepoSpec, set_repo_visibility
from pkgmgr.core.remote_provisioning.visibility import VisibilityOptions
@@ -23,9 +21,7 @@ def _is_git_remote_url(url: str) -> bool:
return True
if u.startswith("ssh://"):
return True
if (u.startswith("https://") or u.startswith("http://")) and u.endswith(".git"):
return True
return False
return bool((u.startswith(("https://", "http://"))) and u.endswith(".git"))
def _provider_hint_from_host(host: str) -> str | None:
@@ -89,7 +85,7 @@ def _print_probe_result(name: str | None, url: str, *, cwd: str) -> None:
def _setup_local_mirrors_for_repo(
repo: Repository,
repositories_base_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
preview: bool,
) -> None:
ctx = build_context(repo, repositories_base_dir, all_repos)
@@ -106,7 +102,7 @@ def _setup_local_mirrors_for_repo(
def _setup_remote_mirrors_for_repo(
repo: Repository,
repositories_base_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
preview: bool,
ensure_remote: bool,
ensure_visibility: str | None,
@@ -195,9 +191,9 @@ def _setup_remote_mirrors_for_repo(
def setup_mirrors(
selected_repos: List[Repository],
selected_repos: list[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
preview: bool = False,
local: bool = True,
remote: bool = True,

6
src/pkgmgr/actions/mirror/types.py
View File

@@ -1,10 +1,10 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict
from typing import Any
Repository = Dict[str, Any]
MirrorMap = Dict[str, str]
Repository = dict[str, Any]
MirrorMap = dict[str, str]
@dataclass(frozen=True)

5
src/pkgmgr/actions/mirror/url_utils.py
View File

@@ -1,11 +1,10 @@
# src/pkgmgr/actions/mirror/url_utils.py
from __future__ import annotations
from typing import Optional, Tuple
from urllib.parse import urlparse
def hostport_from_git_url(url: str) -> Tuple[str, Optional[str]]:
def hostport_from_git_url(url: str) -> tuple[str, str | None]:
url = (url or "").strip()
if not url:
return "", None
@@ -58,7 +57,7 @@ def _strip_dot_git(name: str) -> str:
return n
def parse_repo_from_git_url(url: str) -> Tuple[str, Optional[str], Optional[str]]:
def parse_repo_from_git_url(url: str) -> tuple[str, str | None, str | None]:
"""
Parse (host, owner, repo_name) from common Git remote URLs.

10
src/pkgmgr/actions/mirror/visibility_cmd.py
View File

@@ -1,7 +1,5 @@
from __future__ import annotations
from typing import List
from pkgmgr.core.remote_provisioning import ProviderHint, RepoSpec, set_repo_visibility
from pkgmgr.core.remote_provisioning.visibility import VisibilityOptions
@@ -20,9 +18,7 @@ def _is_git_remote_url(url: str) -> bool:
return True
if u.startswith("ssh://"):
return True
if (u.startswith("https://") or u.startswith("http://")) and u.endswith(".git"):
return True
return False
return bool((u.startswith(("https://", "http://"))) and u.endswith(".git"))
def _provider_hint_from_host(host: str) -> str | None:
@@ -66,9 +62,9 @@ def _apply_visibility_for_url(
def set_mirror_visibility(
selected_repos: List[Repository],
selected_repos: list[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
*,
visibility: str,
preview: bool = False,

2
src/pkgmgr/actions/release/files/__init__.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
"""
Backwards-compatible facade for the release file update helpers.

2
src/pkgmgr/actions/release/files/changelog_lint.py
View File

@@ -1,5 +1,6 @@
from __future__ import annotations
import contextlib
import os
import re
import shutil
@@ -70,6 +71,7 @@ def _markdownlint(reference_path: str, document: str) -> list[str]:
cwd=directory,
capture_output=True,
text=True,
check=False,
)
if proc.returncode == 0:
return []

5
src/pkgmgr/actions/release/files/changelog_md.py
View File

@@ -4,7 +4,6 @@ import os
import re
import sys
from datetime import date
from typing import Optional
from .changelog_lint import (
ChangelogLintError,
@@ -55,7 +54,7 @@ def _insert_after_h1(existing: str, entry: str) -> str:
def update_changelog(
changelog_path: str,
new_version: str,
message: Optional[str] = None,
message: str | None = None,
preview: bool = False,
) -> str:
"""Insert a new release entry into CHANGELOG.md.
@@ -87,7 +86,7 @@ def update_changelog(
elif preview or not sys.stdin.isatty():
body, entry = _entry_for(message or f"Release {new_version}")
else:
attempt: Optional[str] = None
attempt: str | None = None
while True:
print(
"\n[INFO] Provide the changelog entry — a leading '#' becomes "

5
src/pkgmgr/actions/release/files/debian.py
View File

@@ -2,12 +2,11 @@ from __future__ import annotations
import os
from datetime import datetime
from typing import Optional, Tuple
from pkgmgr.core.git.queries import get_config_value
def _get_debian_author() -> Tuple[str, str]:
def _get_debian_author() -> tuple[str, str]:
name = os.environ.get("DEBFULLNAME")
email = os.environ.get("DEBEMAIL")
@@ -33,7 +32,7 @@ def update_debian_changelog(
debian_changelog_path: str,
package_name: str,
new_version: str,
message: Optional[str] = None,
message: str | None = None,
preview: bool = False,
) -> None:
if not os.path.exists(debian_changelog_path):

4
src/pkgmgr/actions/release/files/editor.py
View File

@@ -1,12 +1,12 @@
from __future__ import annotations
import contextlib
import os
import subprocess
import tempfile
from typing import Optional
def _open_editor_for_changelog(initial_message: Optional[str] = None) -> str:
def _open_editor_for_changelog(initial_message: str | None = None) -> str:
editor = os.environ.get("EDITOR", "nano")
with tempfile.NamedTemporaryFile(

2
src/pkgmgr/actions/release/files/pyproject.py
View File

@@ -12,7 +12,7 @@ def update_pyproject_version(
return
try:
with open(pyproject_path, "r", encoding="utf-8") as f:
with open(pyproject_path, encoding="utf-8") as f:
content = f.read()
except OSError as exc:
print(f"[WARN] Could not read pyproject.toml: {exc}")

3
src/pkgmgr/actions/release/files/rpm_changelog.py
View File

@@ -2,7 +2,6 @@ from __future__ import annotations
import os
from datetime import datetime
from typing import Optional
from .debian import _get_debian_author
@@ -11,7 +10,7 @@ def update_spec_changelog(
spec_path: str,
package_name: str,
new_version: str,
message: Optional[str] = None,
message: str | None = None,
preview: bool = False,
) -> None:
if not os.path.exists(spec_path):

7
src/pkgmgr/actions/release/package_name.py
View File

@@ -24,7 +24,6 @@ from __future__ import annotations
import os
import re
from typing import Optional
from pkgmgr.core.repository.paths import RepoPaths
@@ -33,17 +32,17 @@ _PKGBUILD_NAME_RE = re.compile(r"^pkgname=([^\s#]+)\s*$", re.MULTILINE)
_RPM_NAME_RE = re.compile(r"^Name:\s*(\S+)\s*$", re.MULTILINE)
def _read(path: Optional[str]) -> str:
def _read(path: str | None) -> str:
if not path or not os.path.isfile(path):
return ""
try:
with open(path, "r", encoding="utf-8") as f:
with open(path, encoding="utf-8") as f:
return f.read()
except OSError:
return ""
def _extract(pattern: re.Pattern[str], text: str) -> Optional[str]:
def _extract(pattern: re.Pattern[str], text: str) -> str | None:
if not text:
return None
match = pattern.search(text)

2
src/pkgmgr/actions/release/versioning.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
"""
Version discovery and bumping helpers for the release workflow.
"""

7
src/pkgmgr/actions/release/workflow.py
View File

@@ -2,7 +2,6 @@ from __future__ import annotations
import os
import sys
from typing import Optional
from pkgmgr.actions.branch import close_branch
from pkgmgr.core.git import GitRunError, run
@@ -34,7 +33,7 @@ def _release_impl(
pyproject_path: str = "pyproject.toml",
changelog_path: str = "CHANGELOG.md",
release_type: str = "patch",
message: Optional[str] = None,
message: str | None = None,
preview: bool = False,
close: bool = False,
force: bool = False,
@@ -87,7 +86,7 @@ def _release_impl(
else:
print("[INFO] No RPM spec file found. Skipping spec version update.")
effective_message: Optional[str] = message
effective_message: str | None = message
if isinstance(changelog_message, str) and changelog_message.strip():
effective_message = changelog_message.strip()
@@ -195,7 +194,7 @@ def release(
pyproject_path: str = "pyproject.toml",
changelog_path: str = "CHANGELOG.md",
release_type: str = "patch",
message: Optional[str] = None,
message: str | None = None,
preview: bool = False,
force: bool = False,
close: bool = False,

20
src/pkgmgr/actions/repository/_parallel.py
View File

@@ -3,29 +3,29 @@ from __future__ import annotations
import os
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Callable, Dict, List, Tuple
from typing import Any, Callable
from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier
Repository = Dict[str, Any]
RepoRef = Tuple[str, str]
OpResult = Tuple[bool, str]
Repository = dict[str, Any]
RepoRef = tuple[str, str]
OpResult = tuple[bool, str]
RepoOp = Callable[[str], OpResult]
def resolve_repos(
selected_repos: List[Repository],
selected_repos: list[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
) -> List[RepoRef]:
all_repos: list[Repository],
) -> list[RepoRef]:
"""
Resolve ``(identifier, repo_dir)`` pairs for ``selected_repos``.
Repositories whose directory does not exist on disk are reported and
skipped, matching the prior behavior of pull/push handlers.
"""
resolved: List[RepoRef] = []
resolved: list[RepoRef] = []
for repo in selected_repos:
ident = get_repo_identifier(repo, all_repos)
rd = get_repo_dir(repositories_base_dir, repo)
@@ -37,7 +37,7 @@ def resolve_repos(
def run_on_repos(
repos: List[RepoRef],
repos: list[RepoRef],
op: RepoOp,
*,
jobs: int,
@@ -55,7 +55,7 @@ def run_on_repos(
return
effective_jobs = max(1, min(jobs, len(repos)))
failed: List[Tuple[str, str]] = []
failed: list[tuple[str, str]] = []
if effective_jobs == 1:
for ident, rd in repos:

10
src/pkgmgr/actions/repository/clone.py
View File

@@ -1,7 +1,7 @@
from __future__ import annotations
import os
from typing import Any, Dict, List, Optional
from typing import Any
from pkgmgr.core.git.commands import GitCloneError
from pkgmgr.core.git.commands import clone as git_clone
@@ -9,10 +9,10 @@ from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier
from pkgmgr.core.repository.verify import verify_repository
Repository = Dict[str, Any]
Repository = dict[str, Any]
def _build_clone_url(repo: Repository, clone_mode: str) -> Optional[str]:
def _build_clone_url(repo: Repository, clone_mode: str) -> str | None:
provider = repo.get("provider")
account = repo.get("account")
name = repo.get("repository")
@@ -34,9 +34,9 @@ def _build_clone_url(repo: Repository, clone_mode: str) -> Optional[str]:
def clone_repos(
selected_repos: List[Repository],
selected_repos: list[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
all_repos: list[Repository],
preview: bool,
no_verification: bool,
clone_mode: str,

4
src/pkgmgr/actions/repository/create/__init__.py
View File

@@ -1,10 +1,10 @@
from __future__ import annotations
from typing import Any, Dict
from typing import Any
from .service import CreateRepoService
RepositoryConfig = Dict[str, Any]
RepositoryConfig = dict[str, Any]
__all__ = [
"CreateRepoService",

10
src/pkgmgr/actions/repository/create/config_writer.py
View File

@@ -1,21 +1,21 @@
from __future__ import annotations
import os
from typing import Any, Dict, Set
from typing import Any
import yaml
from pkgmgr.core.command.alias import generate_alias
from pkgmgr.core.config.save import save_user_config
Repository = Dict[str, Any]
Repository = dict[str, Any]
class ConfigRepoWriter:
def __init__(
self,
*,
config_merged: Dict[str, Any],
config_merged: dict[str, Any],
user_config_path: str,
bin_dir: str,
):
@@ -43,7 +43,7 @@ class ConfigRepoWriter:
):
return repo
existing_aliases: Set[str] = {
existing_aliases: set[str] = {
str(r.get("alias")) for r in repositories if r.get("alias")
}
@@ -70,7 +70,7 @@ class ConfigRepoWriter:
return repo
if os.path.exists(self.user_config_path):
with open(self.user_config_path, "r", encoding="utf-8") as f:
with open(self.user_config_path, encoding="utf-8") as f:
user_cfg = yaml.safe_load(f) or {}
else:
user_cfg = {}

4
src/pkgmgr/actions/repository/create/mirrors.py
View File

@@ -1,11 +1,11 @@
from __future__ import annotations
from typing import Any, Dict
from typing import Any
from pkgmgr.actions.mirror.io import write_mirrors_file
from pkgmgr.actions.mirror.setup_cmd import setup_mirrors
Repository = Dict[str, Any]
Repository = dict[str, Any]
class MirrorBootstrapper:

3
src/pkgmgr/actions/repository/create/model.py
View File

@@ -1,12 +1,11 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
@dataclass(frozen=True)
class RepoParts:
host: str
port: Optional[str]
port: str | None
owner: str
name: str

3
src/pkgmgr/actions/repository/create/parser.py
View File

@@ -1,7 +1,6 @@
from __future__ import annotations
import re
from typing import Tuple
from urllib.parse import urlparse
from .model import RepoParts
@@ -50,7 +49,7 @@ def _parse_git_url(url: str) -> RepoParts:
return RepoParts(host=host, port=port, owner=owner, name=name)
def _split_host_port(host: str) -> Tuple[str, str | None]:
def _split_host_port(host: str) -> tuple[str, str | None]:
if ":" in host:
h, p = host.split(":", 1)
return h, p or None

4
src/pkgmgr/actions/repository/create/planner.py
View File

@@ -1,7 +1,7 @@
from __future__ import annotations
import os
from typing import Any, Dict
from typing import Any
from .model import RepoParts
@@ -38,7 +38,7 @@ class CreateRepoPlanner:
*,
author_name: str,
author_email: str,
) -> Dict[str, Any]:
) -> dict[str, Any]:
return {
"provider": self.parts.host,
"port": self.parts.port,

4
src/pkgmgr/actions/repository/create/service.py
View File

@@ -1,7 +1,7 @@
from __future__ import annotations
import os
from typing import Any, Dict
from typing import Any
from pkgmgr.core.git.queries import get_config_value
@@ -17,7 +17,7 @@ class CreateRepoService:
def __init__(
self,
*,
config_merged: Dict[str, Any],
config_merged: dict[str, Any],
user_config_path: str,
bin_dir: str,
):

2
src/pkgmgr/actions/repository/create/templates.py
View File

@@ -25,7 +25,7 @@ class TemplateRenderer:
self,
*,
repo_dir: str,
context: Dict[str, Any],
context: dict[str, Any],
preview: bool,
) -> None:
if preview:

36
src/pkgmgr/actions/repository/list.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
"""
Pretty-print repository list with status, categories, tags and path.
@@ -15,9 +13,9 @@ from __future__ import annotations
import os
import re
from textwrap import wrap
from typing import Any, Dict, List, Optional
from typing import Any
Repository = Dict[str, Any]
Repository = dict[str, Any]
RESET = "\033[0m"
BOLD = "\033[1m"
@@ -29,7 +27,7 @@ MAGENTA = "\033[35m"
GREY = "\033[90m"
def _compile_maybe_regex(pattern: str) -> Optional[re.Pattern[str]]:
def _compile_maybe_regex(pattern: str) -> re.Pattern[str] | None:
"""
If pattern is of the form /.../, return a compiled regex (case-insensitive).
Otherwise return None.
@@ -88,7 +86,7 @@ def _compute_status(
"""
Compute a human-readable status string, e.g. 'present,alias,ignored'.
"""
parts: List[str] = []
parts: list[str] = []
exists = os.path.isdir(repo_dir)
if exists:
@@ -128,7 +126,7 @@ def _color_status(status_padded: str) -> str:
pad_spaces = len(status_padded) - len(core)
plain_parts = core.split(",") if core else []
colored_parts: List[str] = []
colored_parts: list[str] = []
for raw_part in plain_parts:
name = raw_part.strip()
@@ -156,12 +154,12 @@ def _color_status(status_padded: str) -> str:
def list_repositories(
repositories: List[Repository],
repositories: list[Repository],
repositories_base_dir: str,
binaries_dir: str,
search_filter: str = "",
status_filter: str = "",
extra_tags: Optional[List[str]] = None,
extra_tags: list[str] | None = None,
show_description: bool = False,
) -> None:
"""
@@ -188,7 +186,7 @@ def list_repositories(
extra_tags = []
search_regex = _compile_maybe_regex(search_filter)
rows: List[Dict[str, Any]] = []
rows: list[dict[str, Any]] = []
# ------------------------------------------------------------------
# Build rows
@@ -208,17 +206,7 @@ def list_repositories(
continue
if search_filter:
haystack = " ".join(
[
identifier,
alias,
provider,
account,
description,
homepage,
repo_dir,
]
)
haystack = f"{identifier} {alias} {provider} {account} {description} {homepage} {repo_dir}"
if search_regex:
if not search_regex.search(haystack):
continue
@@ -226,13 +214,13 @@ def list_repositories(
if search_filter.lower() not in haystack.lower():
continue
categories: List[str] = []
categories: list[str] = []
categories.extend(map(str, repo.get("category_files", [])))
if repo.get("category"):
categories.append(str(repo["category"]))
yaml_tags: List[str] = list(map(str, repo.get("tags", [])))
display_tags: List[str] = sorted(set(yaml_tags + list(map(str, extra_tags))))
yaml_tags: list[str] = list(map(str, repo.get("tags", [])))
display_tags: list[str] = sorted(set(yaml_tags + list(map(str, extra_tags))))
rows.append(
{

22
src/pkgmgr/actions/repository/pull.py
View File

@@ -2,7 +2,7 @@ from __future__ import annotations
import os
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Dict, List, Tuple
from typing import Any
from pkgmgr.actions.repository._parallel import RepoRef, run_on_repos
from pkgmgr.core.git.commands import GitPullArgsError, pull_args
@@ -10,10 +10,10 @@ from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier
from pkgmgr.core.repository.verify import verify_repository
Repository = Dict[str, Any]
Repository = dict[str, Any]
def _pull_one(repo_dir: str, extra_args: List[str], preview: bool) -> Tuple[bool, str]:
def _pull_one(repo_dir: str, extra_args: list[str], preview: bool) -> tuple[bool, str]:
try:
pull_args(extra_args, cwd=repo_dir, preview=preview)
return (True, "")
@@ -25,7 +25,7 @@ def _verify_one(
repo: Repository,
repo_dir: str,
no_verification: bool,
) -> Tuple[bool, bool, List[str]]:
) -> tuple[bool, bool, list[str]]:
"""Returns (has_verified_info, verified_ok, errors)."""
verified_ok, errors, _commit, _key = verify_repository(
repo,
@@ -37,10 +37,10 @@ def _verify_one(
def _verify_all(
candidates: List[Tuple[Repository, str, str]],
candidates: list[tuple[Repository, str, str]],
no_verification: bool,
jobs: int,
) -> List[Tuple[str, str, bool, bool, List[str]]]:
) -> list[tuple[str, str, bool, bool, list[str]]]:
"""
Verify all candidates (parallel if ``jobs > 1``), preserving input order.
@@ -63,10 +63,10 @@ def _verify_all(
def pull_with_verification(
selected_repos: List[Repository],
selected_repos: list[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
extra_args: List[str],
all_repos: list[Repository],
extra_args: list[str],
no_verification: bool,
preview: bool,
jobs: int = 1,
@@ -80,7 +80,7 @@ def pull_with_verification(
- Approved repos are then pulled in parallel when ``jobs > 1``.
- On any pull failure, prints a summary and exits with status 1.
"""
candidates: List[Tuple[Repository, str, str]] = []
candidates: list[tuple[Repository, str, str]] = []
for repo in selected_repos:
ident = get_repo_identifier(repo, all_repos)
rd = get_repo_dir(repositories_base_dir, repo)
@@ -94,7 +94,7 @@ def pull_with_verification(
verify_results = _verify_all(candidates, no_verification, jobs)
approved: List[RepoRef] = []
approved: list[RepoRef] = []
for ident, rd, has_verified_info, verified_ok, errors in verify_results:
if (
not preview

12
src/pkgmgr/actions/repository/push.py
View File

@@ -1,6 +1,6 @@
from __future__ import annotations
from typing import Any, Dict, List, Tuple
from typing import Any
from pkgmgr.actions.repository._parallel import (
resolve_repos,
@@ -8,10 +8,10 @@ from pkgmgr.actions.repository._parallel import (
)
from pkgmgr.core.git.commands import GitPushArgsError, push_args
Repository = Dict[str, Any]
Repository = dict[str, Any]
def _push_one(repo_dir: str, extra_args: List[str], preview: bool) -> Tuple[bool, str]:
def _push_one(repo_dir: str, extra_args: list[str], preview: bool) -> tuple[bool, str]:
try:
push_args(extra_args, cwd=repo_dir, preview=preview)
return (True, "")
@@ -20,10 +20,10 @@ def _push_one(repo_dir: str, extra_args: List[str], preview: bool) -> Tuple[bool
def push_in_parallel(
selected_repos: List[Repository],
selected_repos: list[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
extra_args: List[str],
all_repos: list[Repository],
extra_args: list[str],
preview: bool,
jobs: int = 1,
) -> None:

2
src/pkgmgr/actions/update/__init__.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
from __future__ import annotations
from pkgmgr.actions.update.manager import UpdateManager

6
src/pkgmgr/actions/update/manager.py
View File

@@ -1,9 +1,7 @@
#!/usr/bin/env python3
from __future__ import annotations
from collections.abc import Iterable
from typing import Any, List, Tuple
from typing import Any
from pkgmgr.actions.update.system_updater import SystemUpdater
@@ -37,7 +35,7 @@ class UpdateManager:
from pkgmgr.actions.repository.pull import pull_with_verification
from pkgmgr.core.repository.identifier import get_repo_identifier
failures: List[Tuple[str, str]] = []
failures: list[tuple[str, str]] = []
for repo in list(selected_repos):
identifier = get_repo_identifier(repo, all_repos)

9
src/pkgmgr/actions/update/os_release.py
View File

@@ -1,21 +1,18 @@
#!/usr/bin/env python3
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Dict
def read_os_release(path: str = "/etc/os-release") -> Dict[str, str]:
def read_os_release(path: str = "/etc/os-release") -> dict[str, str]:
"""
Parse /etc/os-release into a dict. Returns empty dict if missing.
"""
if not os.path.exists(path):
return {}
result: Dict[str, str] = {}
with open(path, "r", encoding="utf-8") as f:
result: dict[str, str] = {}
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:

2
src/pkgmgr/actions/update/system_updater.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
from __future__ import annotations
import platform

2
src/pkgmgr/cli/commands/changelog.py
View File

@@ -64,7 +64,7 @@ def _find_previous_and_current_tag(
def handle_changelog(
args,
ctx: CLIContext,
selected: List[Repository],
selected: list[Repository],
) -> None:
"""
Handle the 'changelog' command.

12
src/pkgmgr/cli/commands/config.py
View File

@@ -1,6 +1,4 @@
# src/pkgmgr/cli/commands/config.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
@@ -8,7 +6,7 @@ import os
import shutil
import sys
from pathlib import Path
from typing import Any, Dict, Optional
from typing import Any
import yaml
@@ -21,7 +19,7 @@ from pkgmgr.core.config.save import save_user_config
from pkgmgr.core.repository.resolve import resolve_repos
def _load_user_config(user_config_path: str) -> Dict[str, Any]:
def _load_user_config(user_config_path: str) -> dict[str, Any]:
"""
Load the user config from ~/.config/pkgmgr/config.yaml
(or whatever ctx.user_config_path is), creating the directory if needed.
@@ -32,12 +30,12 @@ def _load_user_config(user_config_path: str) -> Dict[str, Any]:
os.makedirs(cfg_dir, exist_ok=True)
if os.path.exists(user_config_path_expanded):
with open(user_config_path_expanded, "r", encoding="utf-8") as f:
with open(user_config_path_expanded, encoding="utf-8") as f:
return yaml.safe_load(f) or {"repositories": []}
return {"repositories": []}
def _find_defaults_source_dir() -> Optional[str]:
def _find_defaults_source_dir() -> str | None:
"""
Find the directory inside the installed pkgmgr package that contains
the default config files.
@@ -75,7 +73,7 @@ def _update_default_configs(user_config_path: str) -> None:
for name in os.listdir(source_dir):
lower = name.lower()
if not (lower.endswith(".yml") or lower.endswith(".yaml")):
if not (lower.endswith((".yml", ".yaml"))):
continue
if name == "config.yaml":
continue

6
src/pkgmgr/cli/commands/make.py
View File

@@ -1,18 +1,18 @@
from __future__ import annotations
import sys
from typing import Any, Dict, List
from typing import Any
from pkgmgr.actions.proxy import exec_proxy_command
from pkgmgr.cli.context import CLIContext
Repository = Dict[str, Any]
Repository = dict[str, Any]
def handle_make(
args,
ctx: CLIContext,
selected: List[Repository],
selected: list[Repository],
) -> None:
"""
Handle the 'make' command by delegating to exec_proxy_command.

6
src/pkgmgr/cli/commands/mirror.py
View File

@@ -2,7 +2,7 @@
from __future__ import annotations
import sys
from typing import Any, Dict, List
from typing import Any
from pkgmgr.actions.mirror import (
diff_mirrors,
@@ -13,13 +13,13 @@ from pkgmgr.actions.mirror import (
)
from pkgmgr.cli.context import CLIContext
Repository = Dict[str, Any]
Repository = dict[str, Any]
def handle_mirror_command(
ctx: CLIContext,
args: Any,
selected: List[Repository],
selected: list[Repository],
) -> None:
"""
Entry point for 'pkgmgr mirror' subcommands.

6
src/pkgmgr/cli/commands/publish.py
View File

@@ -1,17 +1,17 @@
from __future__ import annotations
import os
from typing import Any, Dict, List
from typing import Any
from pkgmgr.actions.publish import publish
from pkgmgr.cli.context import CLIContext
from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier
Repository = Dict[str, Any]
Repository = dict[str, Any]
def handle_publish(args, ctx: CLIContext, selected: List[Repository]) -> None:
def handle_publish(args, ctx: CLIContext, selected: list[Repository]) -> None:
if not selected:
print("[pkgmgr] No repositories selected for publish.")
return

8
src/pkgmgr/cli/commands/release.py
View File

@@ -1,10 +1,8 @@
#!/usr/bin/env python3
from __future__ import annotations
import os
import sys
from typing import Any, Dict, List
from typing import Any
from pkgmgr.actions.publish import publish as run_publish
from pkgmgr.actions.release import release as run_release
@@ -12,13 +10,13 @@ from pkgmgr.cli.context import CLIContext
from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier
Repository = Dict[str, Any]
Repository = dict[str, Any]
def handle_release(
args,
ctx: CLIContext,
selected: List[Repository],
selected: list[Repository],
) -> None:
if not selected:
print("[pkgmgr] No repositories selected for release.")

8
src/pkgmgr/cli/commands/repos.py
View File

@@ -1,9 +1,7 @@
#!/usr/bin/env python3
from __future__ import annotations
import sys
from typing import Any, Dict, List
from typing import Any
from pkgmgr.actions.install import install_repos
from pkgmgr.actions.repository.create import create_repo
@@ -15,7 +13,7 @@ from pkgmgr.cli.context import CLIContext
from pkgmgr.core.command.run import run_command
from pkgmgr.core.repository.dir import get_repo_dir
Repository = Dict[str, Any]
Repository = dict[str, Any]
def _resolve_repository_directory(repository: Repository, ctx: CLIContext) -> str:
@@ -45,7 +43,7 @@ def _resolve_repository_directory(repository: Repository, ctx: CLIContext) -> st
def handle_repos_command(
args,
ctx: CLIContext,
selected: List[Repository],
selected: list[Repository],
) -> None:
"""
Handle core repository commands (install/update/deinstall/delete/status/list/path/shell/create).

6
src/pkgmgr/cli/commands/tools.py
View File

@@ -1,19 +1,19 @@
from __future__ import annotations
from typing import Any, Dict, List
from typing import Any
from pkgmgr.cli.context import CLIContext
from pkgmgr.cli.tools import open_vscode_workspace
from pkgmgr.cli.tools.paths import resolve_repository_path
from pkgmgr.core.command.run import run_command
Repository = Dict[str, Any]
Repository = dict[str, Any]
def handle_tools_command(
args,
ctx: CLIContext,
selected: List[Repository],
selected: list[Repository],
) -> None:
# ------------------------------------------------------------------
# nautilus "explore" command

4
src/pkgmgr/cli/commands/version.py
View File

@@ -23,7 +23,7 @@ from pkgmgr.core.version.source import (
read_spec_version,
)
Repository = Dict[str, Any]
Repository = dict[str, Any]
def _print_pkgmgr_self_version() -> None:
@@ -74,7 +74,7 @@ def _print_pkgmgr_self_version() -> None:
def handle_version(
args,
ctx: CLIContext,
selected: List[Repository],
selected: list[Repository],
) -> None:
"""
Handle the 'version' command.

6
src/pkgmgr/cli/context.py
View File

@@ -1,7 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict, List
from typing import Any
@dataclass
@@ -13,8 +13,8 @@ class CLIContext:
keeps the CLI layer thin and structured.
"""
config_merged: Dict[str, Any]
config_merged: dict[str, Any]
repositories_base_dir: str
all_repositories: List[Dict[str, Any]]
all_repositories: list[dict[str, Any]]
binaries_dir: str
user_config_path: str

4
src/pkgmgr/cli/dispatch.py
View File

@@ -2,7 +2,7 @@ from __future__ import annotations
import os
import sys
from typing import Any, Dict, List
from typing import Any
from pkgmgr.cli.commands import (
handle_archive,
@@ -34,7 +34,7 @@ def _has_explicit_selection(args) -> bool:
)
def _select_repo_for_current_directory(ctx: CLIContext) -> List[Dict[str, Any]]:
def _select_repo_for_current_directory(ctx: CLIContext) -> list[dict[str, Any]]:
cwd = os.path.abspath(os.getcwd())
matches = []

2
src/pkgmgr/cli/parser/branch_cmd.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse

2
src/pkgmgr/cli/parser/changelog_cmd.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse

5
src/pkgmgr/cli/parser/common.py
View File

@@ -2,7 +2,6 @@
from __future__ import annotations
import argparse
from typing import Optional, Tuple
class SortedSubParsersAction(argparse._SubParsersAction):
@@ -19,8 +18,8 @@ class SortedSubParsersAction(argparse._SubParsersAction):
def _has_action(
parser: argparse.ArgumentParser,
*,
positional: Optional[str] = None,
options: Tuple[str, ...] = (),
positional: str | None = None,
options: tuple[str, ...] = (),
) -> bool:
"""
Check whether the parser already has an action.

2
src/pkgmgr/cli/parser/config_cmd.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse

2
src/pkgmgr/cli/parser/install_update.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
import argparse
from pkgmgr.cli.parser.common import (

2
src/pkgmgr/cli/parser/list_cmd.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse

2
src/pkgmgr/cli/parser/make_cmd.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse

2
src/pkgmgr/cli/parser/mirror_cmd.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse

2
src/pkgmgr/cli/parser/navigation_cmd.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse

2
src/pkgmgr/cli/parser/release_cmd.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse

2
src/pkgmgr/cli/parser/version_cmd.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse

6
src/pkgmgr/cli/proxy.py
View File

@@ -1,11 +1,9 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import os
import sys
from typing import Any, Dict, List
from typing import Any
from pkgmgr.actions.proxy import exec_proxy_command
from pkgmgr.actions.repository.clone import clone_repos
@@ -15,7 +13,7 @@ from pkgmgr.cli.context import CLIContext
from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.selected import get_selected_repos
PROXY_COMMANDS: Dict[str, List[str]] = {
PROXY_COMMANDS: dict[str, list[str]] = {
"git": [
"pull",
"push",

4
src/pkgmgr/cli/tools/paths.py
View File

@@ -1,11 +1,11 @@
from __future__ import annotations
from typing import Any, Dict
from typing import Any
from pkgmgr.cli.context import CLIContext
from pkgmgr.core.repository.dir import get_repo_dir
Repository = Dict[str, Any]
Repository = dict[str, Any]
def resolve_repository_path(repository: Repository, ctx: CLIContext) -> str:

14
src/pkgmgr/cli/tools/vscode.py
View File

@@ -3,14 +3,14 @@ from __future__ import annotations
import json
import os
import shutil
from typing import Any, Dict, List
from typing import Any
from pkgmgr.cli.context import CLIContext
from pkgmgr.cli.tools.paths import resolve_repository_path
from pkgmgr.core.command.run import run_command
from pkgmgr.core.repository.identifier import get_repo_identifier
Repository = Dict[str, Any]
Repository = dict[str, Any]
def _ensure_vscode_cli_available() -> None:
@@ -26,7 +26,7 @@ def _ensure_vscode_cli_available() -> None:
)
def _ensure_identifiers_are_filename_safe(identifiers: List[str]) -> None:
def _ensure_identifiers_are_filename_safe(identifiers: list[str]) -> None:
"""
Ensure identifiers can be used in a filename.
@@ -52,14 +52,14 @@ def _resolve_workspaces_dir(ctx: CLIContext) -> str:
return os.path.expanduser(directories_cfg.get("workspaces", "~/Workspaces"))
def _build_workspace_filename(identifiers: List[str]) -> str:
def _build_workspace_filename(identifiers: list[str]) -> str:
sorted_identifiers = sorted(identifiers)
return "_".join(sorted_identifiers) + ".code-workspace"
def _build_workspace_data(
selected: List[Repository], ctx: CLIContext
) -> Dict[str, Any]:
selected: list[Repository], ctx: CLIContext
) -> dict[str, Any]:
folders = [{"path": resolve_repository_path(repo, ctx)} for repo in selected]
return {
"folders": folders,
@@ -67,7 +67,7 @@ def _build_workspace_data(
}
def open_vscode_workspace(ctx: CLIContext, selected: List[Repository]) -> None:
def open_vscode_workspace(ctx: CLIContext, selected: list[Repository]) -> None:
"""
Create (if missing) and open a VS Code workspace for the selected repositories.

2
src/pkgmgr/core/command/ink.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
import os
from pkgmgr.core.repository.dir import get_repo_dir

22
src/pkgmgr/core/command/resolve.py
View File

@@ -1,15 +1,17 @@
from __future__ import annotations
import os
import shutil
from typing import Any, Dict, List, Optional
from typing import Any
Repository = Dict[str, Any]
Repository = dict[str, Any]
def _is_executable(path: str) -> bool:
return os.path.exists(path) and os.access(path, os.X_OK)
def _find_python_package_root(repo_dir: str) -> Optional[str]:
def _find_python_package_root(repo_dir: str) -> str | None:
"""
Detect a Python src-layout package:
@@ -29,19 +31,19 @@ def _find_python_package_root(repo_dir: str) -> Optional[str]:
return None
def _nix_binary_candidates(home: str, names: List[str]) -> List[str]:
def _nix_binary_candidates(home: str, names: list[str]) -> list[str]:
"""
Build possible Nix profile binary paths for a list of candidate names.
"""
return [os.path.join(home, ".nix-profile", "bin", name) for name in names if name]
def _path_binary_candidates(names: List[str]) -> List[str]:
def _path_binary_candidates(names: list[str]) -> list[str]:
"""
Resolve candidate names via PATH using shutil.which.
Returns only existing, executable paths.
"""
binaries: List[str] = []
binaries: list[str] = []
for name in names:
if not name:
continue
@@ -55,7 +57,7 @@ def resolve_command_for_repo(
repo: Repository,
repo_identifier: str,
repo_dir: str,
) -> Optional[str]:
) -> str | None:
"""
Resolve the executable command for a repository.
@@ -110,7 +112,7 @@ def resolve_command_for_repo(
else:
python_package_name = None
candidate_names: List[str] = []
candidate_names: list[str] = []
seen: set[str] = set()
for name in (
@@ -130,8 +132,8 @@ def resolve_command_for_repo(
path_binaries = _path_binary_candidates(candidate_names)
# b) Classify system (/usr/...) vs non-system
system_binary: Optional[str] = None
non_system_binary: Optional[str] = None
system_binary: str | None = None
non_system_binary: str | None = None
for bin_path in path_binaries:
if bin_path.startswith("/usr"):

11
src/pkgmgr/core/command/run.py
View File

@@ -1,16 +1,17 @@
from __future__ import annotations
import contextlib
import selectors
import subprocess
import sys
from typing import List, Optional, Union
from typing import Union
CommandType = Union[str, List[str]]
CommandType = Union[str, list[str]]
def run_command(
cmd: CommandType,
cwd: Optional[str] = None,
cwd: str | None = None,
preview: bool = False,
allow_failure: bool = False,
) -> subprocess.CompletedProcess:
@@ -48,8 +49,8 @@ def run_command(
sel.register(process.stdout, selectors.EVENT_READ, data="stdout")
sel.register(process.stderr, selectors.EVENT_READ, data="stderr")
stdout_lines: List[str] = []
stderr_lines: List[str] = []
stdout_lines: list[str] = []
stderr_lines: list[str] = []
try:
while sel.get_map():

36
src/pkgmgr/core/config/load.py
View File

@@ -1,6 +1,4 @@
# src/pkgmgr/core/config/load.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Load and merge pkgmgr configuration.
@@ -38,11 +36,11 @@ from __future__ import annotations
import os
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from typing import Any
import yaml
Repo = Dict[str, Any]
Repo = dict[str, Any]
# ---------------------------------------------------------------------------
@@ -50,7 +48,7 @@ Repo = Dict[str, Any]
# ---------------------------------------------------------------------------
def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
"""
Recursively merge two dictionaries.
@@ -64,7 +62,7 @@ def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any
return base
def _repo_key(repo: Repo) -> Tuple[str, str, str]:
def _repo_key(repo: Repo) -> tuple[str, str, str]:
"""
Normalised key for identifying a repository across config files.
"""
@@ -76,10 +74,10 @@ def _repo_key(repo: Repo) -> Tuple[str, str, str]:
def _merge_repo_lists(
base_list: List[Repo],
new_list: List[Repo],
category_name: Optional[str] = None,
) -> List[Repo]:
base_list: list[Repo],
new_list: list[Repo],
category_name: str | None = None,
) -> list[Repo]:
"""
Merge two repository lists, matching by (provider, account, repository).
@@ -87,7 +85,7 @@ def _merge_repo_lists(
- If it exists, its fields are deep-merged (override wins).
- If category_name is set, it is appended to repo["category_files"].
"""
index: Dict[Tuple[str, str, str], Repo] = {_repo_key(r): r for r in base_list}
index: dict[tuple[str, str, str], Repo] = {_repo_key(r): r for r in base_list}
for src in new_list:
key = _repo_key(src)
@@ -120,7 +118,7 @@ def _merge_repo_lists(
return base_list
def _load_yaml_file(path: Path) -> Dict[str, Any]:
def _load_yaml_file(path: Path) -> dict[str, Any]:
"""
Load a single YAML file as dict. Non-dicts yield {}.
"""
@@ -135,8 +133,8 @@ def _load_yaml_file(path: Path) -> Dict[str, Any]:
def _load_layer_dir(
config_dir: Path,
skip_filename: Optional[str] = None,
) -> Dict[str, Any]:
skip_filename: str | None = None,
) -> dict[str, Any]:
"""
Load all *.yml/*.yaml from a directory as layered defaults.
@@ -148,7 +146,7 @@ def _load_layer_dir(
"repositories": [...],
}
"""
defaults: Dict[str, Any] = {"directories": {}, "repositories": []}
defaults: dict[str, Any] = {"directories": {}, "repositories": []}
if not config_dir.is_dir():
return defaults
@@ -186,7 +184,7 @@ def _load_layer_dir(
return defaults
def _load_defaults_from_package_or_project() -> Dict[str, Any]:
def _load_defaults_from_package_or_project() -> dict[str, Any]:
"""
Fallback: load default configs from possible install or dev layouts.
@@ -224,7 +222,7 @@ def _load_defaults_from_package_or_project() -> Dict[str, Any]:
# ---------------------------------------------------------------------------
def load_config(user_config_path: str) -> Dict[str, Any]:
def load_config(user_config_path: str) -> dict[str, Any]:
"""
Load and merge configuration for pkgmgr.
@@ -258,14 +256,14 @@ def load_config(user_config_path: str) -> Dict[str, Any]:
defaults.setdefault("repositories", [])
# 4) User config
user_cfg: Dict[str, Any] = {}
user_cfg: dict[str, Any] = {}
if user_cfg_path.is_file():
user_cfg = _load_yaml_file(user_cfg_path)
user_cfg.setdefault("directories", {})
user_cfg.setdefault("repositories", [])
# 5) Merge
merged: Dict[str, Any] = {}
merged: dict[str, Any] = {}
merged["directories"] = {}
_deep_merge(merged["directories"], defaults["directories"])

3
src/pkgmgr/core/credentials/providers/env.py
View File

@@ -3,7 +3,6 @@ from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Optional
from ..store_keys import env_var_candidates
from ..types import TokenRequest, TokenResult
@@ -15,7 +14,7 @@ class EnvTokenProvider:
source_name: str = "env"
def get(self, request: TokenRequest) -> Optional[TokenResult]:
def get(self, request: TokenRequest) -> TokenResult | None:
for key in env_var_candidates(
request.provider_kind, request.host, request.owner
):

3
src/pkgmgr/core/credentials/providers/gh.py
View File

@@ -3,7 +3,6 @@ from __future__ import annotations
import shutil
import subprocess
from dataclasses import dataclass
from typing import Optional
from ..types import TokenRequest, TokenResult
@@ -18,7 +17,7 @@ class GhTokenProvider:
source_name: str = "gh"
def get(self, request: TokenRequest) -> Optional[TokenResult]:
def get(self, request: TokenRequest) -> TokenResult | None:
# Only meaningful for GitHub-like providers
kind = (request.provider_kind or "").strip().lower()
if kind not in ("github", "github.com"):

5
src/pkgmgr/core/credentials/providers/prompt.py
View File

@@ -4,12 +4,11 @@ from __future__ import annotations
import sys
from dataclasses import dataclass
from getpass import getpass
from typing import Optional
from ..types import TokenRequest, TokenResult
def _token_help_url(provider_kind: str, host: str) -> Optional[str]:
def _token_help_url(provider_kind: str, host: str) -> str | None:
"""
Return a provider-specific URL where a user can create/get an API token.
@@ -51,7 +50,7 @@ class PromptTokenProvider:
source_name: str = "prompt"
def get(self, request: TokenRequest) -> Optional[TokenResult]:
def get(self, request: TokenRequest) -> TokenResult | None:
if not sys.stdin.isatty():
return None

5
src/pkgmgr/core/credentials/store_keys.py
View File

@@ -2,7 +2,6 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
@dataclass(frozen=True)
@@ -14,7 +13,7 @@ class KeyringKey:
def build_keyring_key(
provider_kind: str, host: str, owner: Optional[str]
provider_kind: str, host: str, owner: str | None
) -> KeyringKey:
"""Build a stable keyring key.
@@ -30,7 +29,7 @@ def build_keyring_key(
def env_var_candidates(
provider_kind: str, host: str, owner: Optional[str]
provider_kind: str, host: str, owner: str | None
) -> list[str]:
"""Return a list of environment variable names to try.

Some files were not shown because too many files have changed in this diff Show More