Compare commits

...

12 Commits

Author SHA1 Message Date
Kevin Veen-Birkenbach
ec7b959893 chore(claude): drop Bash(*) and require confirmation for commit and push
Some checks failed
CI / security-codeql (push) Has been cancelled
CI / test-unit (push) Has been cancelled
CI / test-integration (push) Has been cancelled
CI / test-env-virtual (push) Has been cancelled
CI / test-env-nix (push) Has been cancelled
CI / test-e2e (push) Has been cancelled
CI / test-virgin-user (push) Has been cancelled
CI / test-virgin-root (push) Has been cancelled
CI / lint-shell (push) Has been cancelled
CI / lint-python (push) Has been cancelled
CI / lint-docker (push) Has been cancelled
The allow list in this file never took effect. Claude Code discards every
`permissions.allow` entry from project settings while the workspace has
not been trusted (hasTrustDialogAccepted is false for this repo), and it
does so silently in interactive sessions. `Bash(*)` was therefore inert.

Removing it rather than granting trust: the rule only ever bites on the
unsandboxed path, where it would have meant "any command, outside the
sandbox, without a prompt" for everyone who clones this repo. Sandboxed
Bash is auto-allowed anyway, so the entry bought nothing safe.

Add ask rules for git commit and git push instead. Unlike allow rules,
ask and deny are not filtered by the trust gate, and the sandbox
auto-allow path checks them before granting, so they apply to sandboxed
and unsandboxed invocations alike. Both a bare and a wildcard form are
listed because `git commit *` compiles to /^git commit .*$/ and would
not match the bare command.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-28 00:11:29 +02:00
Kevin Veen-Birkenbach
6363709987 fix(code-scanning): stamp report timestamps in UTC
datetime.now() returned a naive timestamp, so `generated_at` in the code
scanning report carried no offset and a reader could not tell which zone
it was written in. Use datetime.now(timezone.utc) for both the report
field and the fallback output directory name.

The changelog keeps date.today(): a release entry carries the author's
local calendar day, so UTC would be wrong there. The rule is suppressed
with that reason instead.

This is the last lint finding; `ruff check src tests` now passes.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-27 18:39:34 +02:00
Kevin Veen-Birkenbach
7764a2946c fix: catch concrete exceptions instead of bare Exception
Replace 46 blind `except Exception` handlers with the exception types the
guarded code can actually raise, so an unforeseen failure surfaces instead
of being silently swallowed. The types were derived per site from the try
block: file reads get (OSError, UnicodeDecodeError), TOML and YAML parsing
add their decode errors, int() guards get ValueError, imports get
ImportError, subprocess calls get (OSError, subprocess.SubprocessError),
git queries get GitRunError, and importlib.metadata lookups get
PackageNotFoundError.

Three handlers stay blind, now with a stated reason: the per-repository
loops in install and update are batch boundaries where one failing
repository must never abort the run.

Three imports come along because the narrowing needs them: GitRunError in
the changelog and version commands, and http.client in token validation --
urllib raises HTTPException, which is not an OSError subclass and would
otherwise have escaped.

Add tests/integration/test_error_path_degradation.py, which drives 44 of
the 46 handlers into their except branch and asserts the documented
degradation. Faults are injected for real rather than mocked wherever
possible: undecodable bytes, malformed YAML, digit strings past
sys.get_int_max_str_digits(), a real HTTPError with an unreadable body.
That approach paid for itself immediately -- tomllib.load() decodes
internally and raises UnicodeDecodeError, which was missing from the
pyproject handlers and would have turned a malformed pyproject.toml into
a traceback.

The two handlers left uncovered are the branch-close path in _release_impl,
reachable only through a full release, and the jinja2 import guard.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-27 18:37:14 +02:00
Kevin Veen-Birkenbach
85d1abf61c fix(credentials): raise a concrete error when the keyring backend fails
The resolver caught bare `Exception` around keyring reads and writes.
That could not be narrowed at the call site: the errors originate in
python-keyring, an optional dependency, so `keyring.errors.KeyringError`
is not referenceable without importing the package unconditionally.

Move the open-ended catch to the boundary where keyring is already
imported. KeyringTokenProvider.get/set now wrap the backend calls and
re-raise as KeyringOperationError, preserving the original exception as
__cause__. The resolver catches that concrete type instead.

KeyringOperationError is deliberately distinct from KeyringUnavailableError:
"no usable backend" and "backend present but the call failed" are different
states, and the resolver treats them differently — the former warns once and
visibly, the latter degrades silently.

Both degradation paths are unchanged: a failed write still lets the token
be used for the current run, a failed read still falls through to the prompt.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-27 16:50:10 +02:00
Kevin Veen-Birkenbach
b4e0594901 style: modernise typing and clean up lint findings
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>
2026-07-27 16:47:13 +02:00
Kevin Veen-Birkenbach
14b95d5639 build(make): add lint target and run it as part of make test
`make lint` runs ruff on src and tests directly on the host, using the
same invocation as the lint-python workflow so local and CI agree. The
script fails with a clear install hint when ruff is not on PATH.

`make test` now depends on `lint`, so linting runs before the container
images are built.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-27 15:25:19 +02:00
Kevin Veen-Birkenbach
8294bce436 style: autoformat src and tests
Repository-wide formatting pass over src/ and tests/: sorted imports and
__all__ entries, dropped redundant `# -*- coding: utf-8 -*-` headers,
unquoted forward-reference annotations (safe under
`from __future__ import annotations`), merged nested `with` statements,
moved `Iterable` and friends from `typing` to `collections.abc`, and
removed `# noqa: F401` markers that no longer suppress anything.

No behavioural changes. Unit and integration suites show the same four
pre-existing failures before and after the pass.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-27 15:17:35 +02:00
Kevin Veen-Birkenbach
f1517963a5 fix(repos create): default new repos to main, not master
`git init` without `-b` inherits the ambient `init.defaultBranch` config,
which is `master` unless the host has configured otherwise, so `pkgmgr
repos create` produced master-based repositories. On top of that, the
push fallback renamed an already-correct `main` branch back to `master`
whenever the initial push failed (missing remote, auth error).

`init()` now requires an explicit `initial_branch` and runs
`git init -b <branch>`; the bootstrapper passes `main` and no longer
falls back to `master` on push failure.

`resolve_base_branch()` keeps its main -> master fallback, since it reads
existing repositories that may legitimately use master.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-27 15:06:09 +02:00
Kevin Veen-Birkenbach
d0e99b6483 Release version 1.16.0 2026-06-28 16:54:16 +02:00
Kevin Veen-Birkenbach
13fc8fe885 feat(cli): add code-scanning command to download GitHub scan results
pkgmgr code-scanning fetches a repository's GitHub code scanning alerts (and analysis metadata) via the gh CLI and writes alerts.json, a readable summary.md digest, and analyses.json into a timestamped directory (default /tmp/<repo>/code-scanner/<YYYYMMDDHHMMSS>) for offline analysis. The repository is resolved from gh or --repo; authentication is delegated to gh; --state filters by alert state.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-28 02:37:20 +02:00
Kevin Veen-Birkenbach
e8aed49a4c feat(release): lint and normalise the changelog message interactively
The release flow now sanitises the release message into the changelog house style and validates it before writing: a leading '#' heading becomes a bold line, inline `code` becomes *italic*, and the resulting entry is checked with markdownlint-cli2 (using the target repo's own config, i.e. the same rules it enforces) or a built-in fallback. In an interactive run the editor re-opens with the findings until the entry is clean; with --message it transforms then aborts on any finding. The editor's ignore-marker moved from '#' to ';' so heading lines survive as content.

debian/changelog and the RPM %changelog mirror the transformed body, keeping all three changelogs consistent.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-28 02:37:19 +02:00
Kevin Veen-Birkenbach
87ee6c66c8 fix(release): preserve multi-line changelog bodies in package mirrors
update_debian_changelog and update_spec_changelog wrapped the whole release message in a single prefixed line, so only the first line of a multi-line (already-bulleted) message was indented/dashed while the rest stayed at column 0. That made dpkg-parsechangelog emit 'badly formatted heading line' and rpmspec report 'bad date', failing the package build. Each non-empty body line is now indented (Debian) or dash-prefixed (RPM) individually.

update_changelog no longer auto-prepends a '* ' bullet to the CHANGELOG.md entry: the message is inserted verbatim with blank-line separation, so a caller-provided markdown list stays markdown-lint-clean instead of gaining a stray leading bullet.

Adds regression tests for the multi-line debian/rpm mirroring and the no-injected-bullet behaviour.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-28 02:37:18 +02:00
283 changed files with 2598 additions and 1425 deletions

View File

@@ -1,11 +1,12 @@
{ {
"permissions": { "permissions": {
"allow": [
"Bash(*)"
],
"ask": [ "ask": [
"Skill(update-config)", "Skill(update-config)",
"Skill(update-config:*)" "Skill(update-config:*)",
"Bash(git commit)",
"Bash(git commit *)",
"Bash(git push)",
"Bash(git push *)"
] ]
}, },
"sandbox": { "sandbox": {

View File

@@ -1,5 +1,11 @@
# Changelog # Changelog
## [1.16.0] - 2026-06-28
* New *code-scanning* command: *pkgmgr code-scanning* downloads a repository's GitHub code scanning alerts and analysis metadata via the *gh* CLI into a timestamped directory (default */tmp/<repo>/code-scanner/<timestamp>*), together with a readable summary, for offline analysis.
* The release flow now lints and normalises the changelog message before writing it: a leading heading becomes bold, inline code becomes italic, and the entry is validated against the repository's markdown rules, re-prompting interactively until it is clean.
* The Debian changelog and the RPM *%changelog* now mirror the full multi-line message correctly — every change line is indented or dash-prefixed — fixing a package build failure where un-indented bodies broke *dpkg-buildpackage* and *rpmspec*; the changelog entry no longer gains an auto-inserted leading bullet.
## [1.15.2] - 2026-05-28 ## [1.15.2] - 2026-05-28
* Restore `infinito` as an alias for the infinito-nexus/core repository so `pkgmgr install infinito` (and friends) resolves again. * Restore `infinito` as an alias for the infinito-nexus/core repository so `pkgmgr install infinito` (and friends) resolves again.

View File

@@ -2,6 +2,7 @@
build build-no-cache build-no-cache-all build-missing \ build build-no-cache build-no-cache-all build-missing \
delete-volumes purge \ delete-volumes purge \
test test-unit test-e2e test-integration test-env-virtual test-env-nix \ test test-unit test-e2e test-integration test-env-virtual test-env-nix \
lint \
setup setup-venv setup-nix setup setup-venv setup-nix
# Distro # Distro
@@ -82,6 +83,13 @@ build-no-cache-all:
PKGMGR_DISTRO="$$d" $(MAKE) build-no-cache; \ PKGMGR_DISTRO="$$d" $(MAKE) build-no-cache; \
done done
# ------------------------------------------------------------
# Lint targets (run on the host, not in a container)
# ------------------------------------------------------------
lint:
@bash scripts/lint/python.sh
# ------------------------------------------------------------ # ------------------------------------------------------------
# Test targets (delegated to scripts/test) # Test targets (delegated to scripts/test)
# ------------------------------------------------------------ # ------------------------------------------------------------
@@ -101,8 +109,8 @@ test-env-virtual: build-missing
test-env-nix: build-missing test-env-nix: build-missing
@bash scripts/test/test-env-nix.sh @bash scripts/test/test-env-nix.sh
# Combined test target for local + CI (unit + integration + e2e) # Combined test target for local + CI (lint + unit + integration + e2e)
test: test-env-virtual test-unit test-integration test-e2e test: lint test-env-virtual test-unit test-integration test-e2e
delete-volumes: delete-volumes:
@docker volume rm "pkgmgr_nix_store_${PKGMGR_DISTRO}" "pkgmgr_nix_cache_${PKGMGR_DISTRO}" || echo "No volumes to delete." @docker volume rm "pkgmgr_nix_store_${PKGMGR_DISTRO}" "pkgmgr_nix_cache_${PKGMGR_DISTRO}" || echo "No volumes to delete."

View File

@@ -32,7 +32,7 @@
rec { rec {
pkgmgr = pyPkgs.buildPythonApplication { pkgmgr = pyPkgs.buildPythonApplication {
pname = "package-manager"; pname = "package-manager";
version = "1.15.2"; version = "1.16.0";
# Use the git repo as source # Use the git repo as source
src = ./.; src = ./.;

View File

@@ -1,7 +1,7 @@
# Maintainer: Kevin Veen-Birkenbach <info@veen.world> # Maintainer: Kevin Veen-Birkenbach <info@veen.world>
pkgname=package-manager pkgname=package-manager
pkgver=1.15.2 pkgver=1.16.0
pkgrel=1 pkgrel=1
pkgdesc="Local-flake wrapper for Kevin's package-manager (Nix-based)." pkgdesc="Local-flake wrapper for Kevin's package-manager (Nix-based)."
arch=('any') arch=('any')

View File

@@ -1,3 +1,11 @@
package-manager (1.16.0-1) unstable; urgency=medium
* New *code-scanning* command: *pkgmgr code-scanning* downloads a repository's GitHub code scanning alerts and analysis metadata via the *gh* CLI into a timestamped directory (default */tmp/<repo>/code-scanner/<timestamp>*), together with a readable summary, for offline analysis.
* The release flow now lints and normalises the changelog message before writing it: a leading heading becomes bold, inline code becomes italic, and the entry is validated against the repository's markdown rules, re-prompting interactively until it is clean.
* The Debian changelog and the RPM *%changelog* now mirror the full multi-line message correctly — every change line is indented or dash-prefixed — fixing a package build failure where un-indented bodies broke *dpkg-buildpackage* and *rpmspec*; the changelog entry no longer gains an auto-inserted leading bullet.
-- Kevin Veen-Birkenbach <kevin@veen.world> Sun, 28 Jun 2026 16:54:16 +0200
package-manager (1.15.2-1) unstable; urgency=medium package-manager (1.15.2-1) unstable; urgency=medium
* Restore `infinito` as an alias for the infinito-nexus/core repository so `pkgmgr install infinito` (and friends) resolves again. * Restore `infinito` as an alias for the infinito-nexus/core repository so `pkgmgr install infinito` (and friends) resolves again.

View File

@@ -1,5 +1,5 @@
Name: package-manager Name: package-manager
Version: 1.15.2 Version: 1.16.0
Release: 1%{?dist} Release: 1%{?dist}
Summary: Wrapper that runs Kevin's package-manager via Nix flake Summary: Wrapper that runs Kevin's package-manager via Nix flake
@@ -74,6 +74,11 @@ echo ">>> package-manager removed. Nix itself was not removed."
/usr/lib/package-manager/ /usr/lib/package-manager/
%changelog %changelog
* Sun Jun 28 2026 Kevin Veen-Birkenbach <kevin@veen.world> - 1.16.0-1
- * New *code-scanning* command: *pkgmgr code-scanning* downloads a repository's GitHub code scanning alerts and analysis metadata via the *gh* CLI into a timestamped directory (default */tmp/<repo>/code-scanner/<timestamp>*), together with a readable summary, for offline analysis.
- * The release flow now lints and normalises the changelog message before writing it: a leading heading becomes bold, inline code becomes italic, and the entry is validated against the repository's markdown rules, re-prompting interactively until it is clean.
- * The Debian changelog and the RPM *%changelog* now mirror the full multi-line message correctly — every change line is indented or dash-prefixed — fixing a package build failure where un-indented bodies broke *dpkg-buildpackage* and *rpmspec*; the changelog entry no longer gains an auto-inserted leading bullet.
* Thu May 28 2026 Kevin Veen-Birkenbach <kevin@veen.world> - 1.15.2-1 * Thu May 28 2026 Kevin Veen-Birkenbach <kevin@veen.world> - 1.15.2-1
- Restore `infinito` as an alias for the infinito-nexus/core repository so `pkgmgr install infinito` (and friends) resolves again. - Restore `infinito` as an alias for the infinito-nexus/core repository so `pkgmgr install infinito` (and friends) resolves again.

View File

@@ -7,7 +7,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "kpmx" name = "kpmx"
version = "1.15.2" version = "1.16.0"
description = "Kevin's package-manager tool (pkgmgr)" description = "Kevin's package-manager tool (pkgmgr)"
readme = "README.md" readme = "README.md"
requires-python = ">=3.9" requires-python = ">=3.9"

15
scripts/lint/python.sh Executable file
View File

@@ -0,0 +1,15 @@
#!/usr/bin/env bash
set -euo pipefail
echo "============================================================"
echo ">>> Running RUFF lint on src and tests (local, no container)"
echo "============================================================"
if ! command -v ruff >/dev/null 2>&1; then
echo "ruff is not installed or not on PATH." >&2
echo "Install it with: pip install ruff" >&2
exit 127
fi
ruff --version
ruff check src tests

View File

@@ -1,6 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" """
Top-level pkgmgr package. Top-level pkgmgr package.

View File

@@ -1,6 +1,6 @@
from __future__ import annotations from __future__ import annotations
# expose subpackages for patch() / resolve_name() friendliness # expose subpackages for patch() / resolve_name() friendliness
from . import release as release # noqa: F401 from . import release as release
__all__ = ["release"] __all__ = ["release"]

View File

@@ -3,8 +3,8 @@
from __future__ import annotations from __future__ import annotations
import re import re
from collections.abc import Iterable
from pathlib import Path from pathlib import Path
from typing import Iterable
DEFAULT_FILENAME_PATTERN = re.compile(r"^\d{3}-[^/]+\.md$") DEFAULT_FILENAME_PATTERN = re.compile(r"^\d{3}-[^/]+\.md$")
TEMPLATE_FILENAME = "000-template.md" TEMPLATE_FILENAME = "000-template.md"

View File

@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import contextlib
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@@ -101,10 +102,8 @@ def run_archive(
if not dry_run: if not dry_run:
for path, _title in archived: for path, _title in archived:
try: with contextlib.suppress(FileNotFoundError):
path.unlink() path.unlink()
except FileNotFoundError:
pass
return ArchivePlan( return ArchivePlan(
archived=archived, archived=archived,

View File

@@ -1,14 +1,13 @@
# -*- coding: utf-8 -*-
""" """
Public API for branch actions. Public API for branch actions.
""" """
from .open_branch import open_branch
from .close_branch import close_branch from .close_branch import close_branch
from .drop_branch import drop_branch from .drop_branch import drop_branch
from .open_branch import open_branch
__all__ = [ __all__ = [
"open_branch",
"close_branch", "close_branch",
"drop_branch", "drop_branch",
"open_branch",
] ]

View File

@@ -1,9 +1,5 @@
from __future__ import annotations from __future__ import annotations
from typing import Optional
from pkgmgr.core.git.errors import GitRunError
from pkgmgr.core.git.queries import get_current_branch
from pkgmgr.core.git.commands import ( from pkgmgr.core.git.commands import (
GitDeleteRemoteBranchError, GitDeleteRemoteBranchError,
checkout, checkout,
@@ -14,12 +10,12 @@ from pkgmgr.core.git.commands import (
pull, pull,
push, push,
) )
from pkgmgr.core.git.errors import GitRunError
from pkgmgr.core.git.queries import resolve_base_branch from pkgmgr.core.git.queries import get_current_branch, resolve_base_branch
def close_branch( def close_branch(
name: Optional[str], name: str | None,
base_branch: str = "main", base_branch: str = "main",
fallback_base: str = "master", fallback_base: str = "master",
cwd: str = ".", cwd: str = ".",

View File

@@ -1,20 +1,16 @@
from __future__ import annotations from __future__ import annotations
from typing import Optional
from pkgmgr.core.git.errors import GitRunError
from pkgmgr.core.git.queries import get_current_branch
from pkgmgr.core.git.commands import ( from pkgmgr.core.git.commands import (
GitDeleteRemoteBranchError, GitDeleteRemoteBranchError,
delete_local_branch, delete_local_branch,
delete_remote_branch, delete_remote_branch,
) )
from pkgmgr.core.git.errors import GitRunError
from pkgmgr.core.git.queries import resolve_base_branch from pkgmgr.core.git.queries import get_current_branch, resolve_base_branch
def drop_branch( def drop_branch(
name: Optional[str], name: str | None,
base_branch: str = "main", base_branch: str = "main",
fallback_base: str = "master", fallback_base: str = "master",
cwd: str = ".", cwd: str = ".",

View File

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

View File

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

View File

@@ -0,0 +1,161 @@
"""Download GitHub code scanning results via the ``gh`` CLI.
Fetches all code scanning alerts (and the analysis metadata) for a
repository and writes them, plus a readable digest, into a timestamped
directory so they can be read and analysed offline. Authentication is
delegated to ``gh`` (credentials from its keyring/login).
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
from collections import Counter
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
class CodeScanningError(RuntimeError):
"""Raised when code scanning results cannot be downloaded."""
@dataclass
class CodeScanningResult:
repo: str
output_dir: str
alert_count: int
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, check=False)
def _resolve_repo(repo: str | None) -> str:
if repo:
return repo
proc = _gh(["repo", "view", "--json", "nameWithOwner", "-q", ".nameWithOwner"])
name = proc.stdout.strip()
if proc.returncode != 0 or not name:
raise CodeScanningError(
"could not resolve the current repository; run inside a GitHub "
"repo or pass --repo OWNER/REPO "
f"({proc.stderr.strip()})"
)
return name
def _fetch_json(endpoint: str, params: list[str] | None = None) -> Any:
args = ["api", endpoint, "--paginate"]
for param in params or []:
args += ["-f", param]
proc = _gh(args)
if proc.returncode != 0:
raise CodeScanningError(
f"gh api {endpoint} failed: {proc.stderr.strip() or 'unknown error'}"
)
body = proc.stdout.strip()
return json.loads(body) if body else []
def _alert_row(alert: dict) -> str:
rule = alert.get("rule") or {}
instance = alert.get("most_recent_instance") or {}
location = instance.get("location") or {}
message = (instance.get("message") or {}).get("text", "").strip().replace("\n", " ")
severity = rule.get("security_severity_level") or rule.get("severity") or "unknown"
path = location.get("path", "?")
line = location.get("start_line", "?")
state = alert.get("state", "?")
rule_id = rule.get("id", "?")
return f"- [{severity}] {rule_id}{path}:{line} ({state})\n {message}"
def _build_summary(repo: str, generated_at: str, alerts: list[dict]) -> str:
by_severity: Counter = Counter()
by_state: Counter = Counter()
by_rule: Counter = Counter()
for alert in alerts:
rule = alert.get("rule") or {}
by_severity[
rule.get("security_severity_level") or rule.get("severity") or "unknown"
] += 1
by_state[alert.get("state", "unknown")] += 1
by_rule[rule.get("id", "unknown")] += 1
lines = [
f"# Code scanning summary — {repo}",
"",
f"- Generated: {generated_at}",
f"- Total alerts: {len(alerts)}",
"",
"## By severity",
"",
]
lines += [f"- {sev}: {count}" for sev, count in by_severity.most_common()] or [
"- none"
]
lines += ["", "## By state", ""]
lines += [f"- {state}: {count}" for state, count in by_state.most_common()] or [
"- none"
]
lines += ["", "## By rule", ""]
lines += [f"- {rule}: {count}" for rule, count in by_rule.most_common()] or [
"- none"
]
lines += ["", "## Alerts", ""]
lines += [_alert_row(alert) for alert in alerts] or ["- none"]
return "\n".join(lines) + "\n"
def download_code_scanning(
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")
repo = _resolve_repo(repo)
repo_name = repo.rstrip("/").split("/")[-1]
if output_dir is None:
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
output_dir = os.path.join("/tmp", repo_name, "code-scanner", timestamp)
output_dir = os.path.abspath(os.path.expanduser(output_dir))
os.makedirs(output_dir, exist_ok=True)
generated_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
alerts = _fetch_json(
f"repos/{repo}/code-scanning/alerts",
params=[f"state={state}"] if state else None,
)
files: list[str] = []
def _write(name: str, content: str) -> None:
path = os.path.join(output_dir, name)
with open(path, "w", encoding="utf-8") as handle:
handle.write(content)
files.append(path)
_write("alerts.json", json.dumps(alerts, indent=2, ensure_ascii=False) + "\n")
_write("summary.md", _build_summary(repo, generated_at, alerts))
try:
analyses = _fetch_json(f"repos/{repo}/code-scanning/analyses")
_write(
"analyses.json", json.dumps(analyses, indent=2, ensure_ascii=False) + "\n"
)
except CodeScanningError as exc:
_write("analyses.json", json.dumps({"error": str(exc)}, indent=2) + "\n")
return CodeScanningResult(
repo=repo,
output_dir=output_dir,
alert_count=len(alerts),
files=files,
)

View File

@@ -1,5 +1,7 @@
import yaml
import os import os
import yaml
from pkgmgr.core.config.save import save_user_config from pkgmgr.core.config.save import save_user_config
@@ -28,7 +30,7 @@ def interactive_add(config, USER_CONFIG_PATH: str):
confirm = input("Add this entry to user config? (y/N): ").strip().lower() confirm = input("Add this entry to user config? (y/N): ").strip().lower()
if confirm == "y": if confirm == "y":
if os.path.exists(USER_CONFIG_PATH): 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 {} user_config = yaml.safe_load(f) or {}
else: else:
user_config = {"repositories": []} user_config = {"repositories": []}

View File

@@ -1,6 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" """
Initialize user configuration by scanning the repositories base directory. Initialize user configuration by scanning the repositories base directory.
@@ -23,7 +20,7 @@ For each discovered repository, the function:
from __future__ import annotations from __future__ import annotations
import os import os
from typing import Any, Dict from typing import Any
from pkgmgr.core.command.alias import generate_alias from pkgmgr.core.command.alias import generate_alias
from pkgmgr.core.config.save import save_user_config from pkgmgr.core.config.save import save_user_config
@@ -31,8 +28,8 @@ from pkgmgr.core.git.queries import get_latest_commit
def config_init( def config_init(
user_config: Dict[str, Any], user_config: dict[str, Any],
defaults_config: Dict[str, Any], defaults_config: dict[str, Any],
bin_dir: str, bin_dir: str,
user_config_path: str, user_config_path: str,
) -> None: ) -> None:
@@ -55,7 +52,7 @@ def config_init(
print("[INIT] Scanning repository base directory:") print("[INIT] Scanning repository base directory:")
print(f" {repositories_base_dir}") print(f" {repositories_base_dir}")
print("") print()
if not os.path.isdir(repositories_base_dir): if not os.path.isdir(repositories_base_dir):
print(f"[ERROR] Base directory does not exist: {repositories_base_dir}") print(f"[ERROR] Base directory does not exist: {repositories_base_dir}")
@@ -129,7 +126,7 @@ def config_init(
"[WARN] Could not read commit (not a git repo or no commits)." "[WARN] Could not read commit (not a git repo or no commits)."
) )
entry: Dict[str, Any] = { entry: dict[str, Any] = {
"provider": provider, "provider": provider,
"account": account, "account": account,
"repository": repo_name, "repository": repo_name,
@@ -153,7 +150,7 @@ def config_init(
new_entries.append(entry) new_entries.append(entry)
print("") # blank line between accounts print() # blank line between accounts
# ------------------------------------------------------------ # ------------------------------------------------------------
# Summary # Summary

View File

@@ -1,4 +1,5 @@
import yaml import yaml
from pkgmgr.core.config.load import load_config from pkgmgr.core.config.load import load_config

View File

@@ -1,6 +1,4 @@
# src/pkgmgr/actions/install/__init__.py # src/pkgmgr/actions/install/__init__.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" """
High-level entry point for repository installation. High-level entry point for repository installation.
@@ -16,28 +14,28 @@ Responsibilities:
from __future__ import annotations from __future__ import annotations
import os import os
from typing import Any, Dict, List, Optional, Tuple from typing import Any
from pkgmgr.core.repository.identifier import get_repo_identifier
from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.verify import verify_repository
from pkgmgr.actions.repository.clone import clone_repos
from pkgmgr.actions.install.context import RepoContext from pkgmgr.actions.install.context import RepoContext
from pkgmgr.actions.install.installers.makefile import (
MakefileInstaller,
)
from pkgmgr.actions.install.installers.nix import (
NixFlakeInstaller,
)
from pkgmgr.actions.install.installers.os_packages import ( from pkgmgr.actions.install.installers.os_packages import (
ArchPkgbuildInstaller, ArchPkgbuildInstaller,
DebianControlInstaller, DebianControlInstaller,
RpmSpecInstaller, RpmSpecInstaller,
) )
from pkgmgr.actions.install.installers.nix import (
NixFlakeInstaller,
)
from pkgmgr.actions.install.installers.python import PythonInstaller from pkgmgr.actions.install.installers.python import PythonInstaller
from pkgmgr.actions.install.installers.makefile import (
MakefileInstaller,
)
from pkgmgr.actions.install.pipeline import InstallationPipeline from pkgmgr.actions.install.pipeline import InstallationPipeline
from pkgmgr.actions.repository.clone import clone_repos
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 = [ INSTALLERS = [
ArchPkgbuildInstaller(), ArchPkgbuildInstaller(),
@@ -52,12 +50,12 @@ INSTALLERS = [
def _ensure_repo_dir( def _ensure_repo_dir(
repo: Repository, repo: Repository,
repositories_base_dir: str, repositories_base_dir: str,
all_repos: List[Repository], all_repos: list[Repository],
preview: bool, preview: bool,
no_verification: bool, no_verification: bool,
clone_mode: str, clone_mode: str,
identifier: str, identifier: str,
) -> Optional[str]: ) -> str | None:
""" """
Compute and, if necessary, clone the repository directory. Compute and, if necessary, clone the repository directory.
@@ -127,7 +125,7 @@ def _create_context(
repo_dir: str, repo_dir: str,
repositories_base_dir: str, repositories_base_dir: str,
bin_dir: str, bin_dir: str,
all_repos: List[Repository], all_repos: list[Repository],
no_verification: bool, no_verification: bool,
preview: bool, preview: bool,
quiet: bool, quiet: bool,
@@ -155,10 +153,10 @@ def _create_context(
def install_repos( def install_repos(
selected_repos: List[Repository], selected_repos: list[Repository],
repositories_base_dir: str, repositories_base_dir: str,
bin_dir: str, bin_dir: str,
all_repos: List[Repository], all_repos: list[Repository],
no_verification: bool, no_verification: bool,
preview: bool, preview: bool,
quiet: bool, quiet: bool,
@@ -179,7 +177,7 @@ def install_repos(
overall command never exits non-zero because of per-repository failures. overall command never exits non-zero because of per-repository failures.
""" """
pipeline = InstallationPipeline(INSTALLERS) pipeline = InstallationPipeline(INSTALLERS)
failures: List[Tuple[str, str]] = [] failures: list[tuple[str, str]] = []
for repo in selected_repos: for repo in selected_repos:
identifier = get_repo_identifier(repo, all_repos) identifier = get_repo_identifier(repo, all_repos)
@@ -232,7 +230,7 @@ def install_repos(
f"[Warning] install: repository {identifier} failed (exit={code}). Continuing..." f"[Warning] install: repository {identifier} failed (exit={code}). Continuing..."
) )
continue continue
except Exception as exc: except Exception as exc: # noqa: BLE001 - batch boundary: one repository must never abort the run
failures.append((identifier, f"unexpected error: {exc}")) failures.append((identifier, f"unexpected error: {exc}"))
if not quiet: if not quiet:
print( print(

View File

@@ -1,6 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" """
Capability detection for pkgmgr. Capability detection for pkgmgr.
@@ -35,7 +32,8 @@ from __future__ import annotations
import glob import glob
import os import os
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from typing import Iterable, TYPE_CHECKING, Optional from collections.abc import Iterable
from typing import TYPE_CHECKING
if TYPE_CHECKING: if TYPE_CHECKING:
from pkgmgr.actions.install.context import RepoContext 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.""" """Read a file as UTF-8 text, returning None if it does not exist or fails."""
if not os.path.exists(path): if not os.path.exists(path):
return None return None
try: try:
with open(path, "r", encoding="utf-8") as f: with open(path, encoding="utf-8") as f:
return f.read() return f.read()
except OSError: except OSError:
return None return None
@@ -75,12 +73,12 @@ def _scan_files_for_patterns(files: Iterable[str], patterns: Iterable[str]) -> b
return False 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.""" """Return the first *.spec file in repo_dir, if any."""
matches = glob.glob(os.path.join(repo_dir, "*.spec")) matches = glob.glob(os.path.join(repo_dir, "*.spec"))
if not matches: if not matches:
return None return None
return sorted(matches)[0] return min(matches)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -100,7 +98,7 @@ class CapabilityMatcher(ABC):
raise NotImplementedError raise NotImplementedError
@abstractmethod @abstractmethod
def is_provided(self, ctx: "RepoContext", layer: str) -> bool: def is_provided(self, ctx: RepoContext, layer: str) -> bool:
""" """
Return True if this capability is actually provided by the given layer Return True if this capability is actually provided by the given layer
for this repository. for this repository.
@@ -133,7 +131,7 @@ class PythonRuntimeCapability(CapabilityMatcher):
# OS packages may wrap Python builds, but must explicitly prove it # OS packages may wrap Python builds, but must explicitly prove it
return layer in {"python", "nix", "os-packages"} return layer in {"python", "nix", "os-packages"}
def is_provided(self, ctx: "RepoContext", layer: str) -> bool: def is_provided(self, ctx: RepoContext, layer: str) -> bool:
repo_dir = ctx.repo_dir repo_dir = ctx.repo_dir
if layer == "python": if layer == "python":
@@ -208,7 +206,7 @@ class MakeInstallCapability(CapabilityMatcher):
def applies_to_layer(self, layer: str) -> bool: def applies_to_layer(self, layer: str) -> bool:
return layer in {"makefile", "python", "nix", "os-packages"} return layer in {"makefile", "python", "nix", "os-packages"}
def is_provided(self, ctx: "RepoContext", layer: str) -> bool: def is_provided(self, ctx: RepoContext, layer: str) -> bool:
repo_dir = ctx.repo_dir repo_dir = ctx.repo_dir
if layer == "makefile": if layer == "makefile":
@@ -216,7 +214,7 @@ class MakeInstallCapability(CapabilityMatcher):
if not os.path.exists(makefile): if not os.path.exists(makefile):
return False return False
try: try:
with open(makefile, "r", encoding="utf-8") as f: with open(makefile, encoding="utf-8") as f:
for line in f: for line in f:
if line.strip().startswith("install:"): if line.strip().startswith("install:"):
return True return True
@@ -274,7 +272,7 @@ class NixFlakeCapability(CapabilityMatcher):
# Only Nix itself and OS packages that explicitly wrap Nix # Only Nix itself and OS packages that explicitly wrap Nix
return layer in {"nix", "os-packages"} return layer in {"nix", "os-packages"}
def is_provided(self, ctx: "RepoContext", layer: str) -> bool: def is_provided(self, ctx: RepoContext, layer: str) -> bool:
repo_dir = ctx.repo_dir repo_dir = ctx.repo_dir
if layer == "nix": if layer == "nix":
@@ -328,7 +326,7 @@ LAYER_ORDER: list[str] = [
def detect_capabilities( def detect_capabilities(
ctx: "RepoContext", ctx: RepoContext,
layers: Iterable[str], layers: Iterable[str],
) -> dict[str, set[str]]: ) -> dict[str, set[str]]:
""" """
@@ -359,8 +357,8 @@ def detect_capabilities(
def resolve_effective_capabilities( def resolve_effective_capabilities(
ctx: "RepoContext", ctx: RepoContext,
layers: Optional[Iterable[str]] = None, layers: Iterable[str] | None = None,
) -> dict[str, set[str]]: ) -> dict[str, set[str]]:
""" """
Resolve *effective* capabilities for each layer using a bottom-up strategy. 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 This means *any* higher layer can overshadow a lower layer, not just
a specific one like Nix. The resolver is completely generic. a specific one like Nix. The resolver is completely generic.
""" """
if layers is None: layers_list = list(LAYER_ORDER) if layers is None else list(layers)
layers_list = list(LAYER_ORDER)
else:
layers_list = list(layers)
raw_caps = detect_capabilities(ctx, layers_list) raw_caps = detect_capabilities(ctx, layers_list)
effective: dict[str, set[str]] = {layer: set() for layer in layers_list} effective: dict[str, set[str]] = {layer: set() for layer in layers_list}

View File

@@ -1,6 +1,4 @@
# src/pkgmgr/actions/install/context.py # src/pkgmgr/actions/install/context.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" """
Shared context object for repository installation steps. 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 dataclasses import dataclass
from typing import Any, Dict, List from typing import Any
@dataclass @dataclass
class RepoContext: class RepoContext:
"""Container for all repository-related data used during installation.""" """Container for all repository-related data used during installation."""
repo: Dict[str, Any] repo: dict[str, Any]
identifier: str identifier: str
repo_dir: str repo_dir: str
repositories_base_dir: str repositories_base_dir: str
bin_dir: str bin_dir: str
all_repos: List[Dict[str, Any]] all_repos: list[dict[str, Any]]
no_verification: bool no_verification: bool
preview: bool preview: bool

View File

@@ -1,6 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" """
Installer package for pkgmgr. Installer package for pkgmgr.
@@ -9,15 +6,17 @@ pkgmgr.actions.install.installers.
""" """
from pkgmgr.actions.install.installers.base import BaseInstaller # noqa: F401 from pkgmgr.actions.install.installers.base import BaseInstaller # noqa: F401
from pkgmgr.actions.install.installers.nix import NixFlakeInstaller # noqa: F401
from pkgmgr.actions.install.installers.python import PythonInstaller # noqa: F401
from pkgmgr.actions.install.installers.makefile import MakefileInstaller # noqa: F401 from pkgmgr.actions.install.installers.makefile import MakefileInstaller # noqa: F401
from pkgmgr.actions.install.installers.nix import NixFlakeInstaller # noqa: F401
# OS-specific installers # OS-specific installers
from pkgmgr.actions.install.installers.os_packages.arch_pkgbuild import ( from pkgmgr.actions.install.installers.os_packages.arch_pkgbuild import (
ArchPkgbuildInstaller as ArchPkgbuildInstaller, ArchPkgbuildInstaller as ArchPkgbuildInstaller,
) # noqa: F401 )
from pkgmgr.actions.install.installers.os_packages.debian_control import ( from pkgmgr.actions.install.installers.os_packages.debian_control import (
DebianControlInstaller as DebianControlInstaller, DebianControlInstaller as DebianControlInstaller,
) # noqa: F401 )
from pkgmgr.actions.install.installers.os_packages.rpm_spec import RpmSpecInstaller # noqa: F401 from pkgmgr.actions.install.installers.os_packages.rpm_spec import (
RpmSpecInstaller, # noqa: F401
)
from pkgmgr.actions.install.installers.python import PythonInstaller # noqa: F401

View File

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

View File

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

View File

@@ -1,6 +1,6 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING, List from typing import TYPE_CHECKING
from .profile import NixProfileInspector from .profile import NixProfileInspector
from .retry import GitHubRateLimitRetry from .retry import GitHubRateLimitRetry
@@ -33,7 +33,7 @@ class NixConflictResolver:
def resolve( def resolve(
self, self,
ctx: "RepoContext", ctx: RepoContext,
install_cmd: str, install_cmd: str,
stdout: str, stdout: str,
stderr: str, stderr: str,
@@ -49,7 +49,7 @@ class NixConflictResolver:
store_prefixes = self._parser.existing_store_prefixes(combined) store_prefixes = self._parser.existing_store_prefixes(combined)
# 2) Resolve them to concrete remove tokens # 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, ctx,
self._runner, self._runner,
store_prefixes, store_prefixes,

View File

@@ -2,7 +2,7 @@ from __future__ import annotations
import os import os
import shutil import shutil
from typing import TYPE_CHECKING, List, Tuple from typing import TYPE_CHECKING
from pkgmgr.actions.install.installers.base import BaseInstaller from pkgmgr.actions.install.installers.base import BaseInstaller
@@ -28,7 +28,7 @@ class NixFlakeInstaller(BaseInstaller):
# Newer nix rejects numeric indices; we learn this at runtime and cache the decision. # Newer nix rejects numeric indices; we learn this at runtime and cache the decision.
self._indices_supported: bool | None = None self._indices_supported: bool | None = None
def supports(self, ctx: "RepoContext") -> bool: def supports(self, ctx: RepoContext) -> bool:
if os.environ.get("PKGMGR_DISABLE_NIX_FLAKE_INSTALLER") == "1": if os.environ.get("PKGMGR_DISABLE_NIX_FLAKE_INSTALLER") == "1":
if not ctx.quiet: if not ctx.quiet:
print( print(
@@ -42,13 +42,13 @@ class NixFlakeInstaller(BaseInstaller):
return os.path.exists(os.path.join(ctx.repo_dir, self.FLAKE_FILE)) 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) # (output_name, allow_failure)
if ctx.identifier in {"pkgmgr", "package-manager"}: if ctx.identifier in {"pkgmgr", "package-manager"}:
return [("pkgmgr", False), ("default", True)] return [("pkgmgr", False), ("default", True)]
return [("default", False)] return [("default", False)]
def run(self, ctx: "RepoContext") -> None: def run(self, ctx: RepoContext) -> None:
if not self.supports(ctx): if not self.supports(ctx):
return return
@@ -68,7 +68,7 @@ class NixFlakeInstaller(BaseInstaller):
else: else:
self._install_only(ctx, output, allow_failure) self._install_only(ctx, output, allow_failure)
def _installable(self, ctx: "RepoContext", output: str) -> str: def _installable(self, ctx: RepoContext, output: str) -> str:
return f"{ctx.repo_dir}#{output}" return f"{ctx.repo_dir}#{output}"
# --------------------------------------------------------------------- # ---------------------------------------------------------------------
@@ -76,7 +76,7 @@ class NixFlakeInstaller(BaseInstaller):
# --------------------------------------------------------------------- # ---------------------------------------------------------------------
def _install_only( def _install_only(
self, ctx: "RepoContext", output: str, allow_failure: bool self, ctx: RepoContext, output: str, allow_failure: bool
) -> None: ) -> None:
install_cmd = f"nix profile install {self._installable(ctx, output)}" install_cmd = f"nix profile install {self._installable(ctx, output)}"
@@ -162,7 +162,7 @@ class NixFlakeInstaller(BaseInstaller):
# --------------------------------------------------------------------- # ---------------------------------------------------------------------
def _force_upgrade_output( def _force_upgrade_output(
self, ctx: "RepoContext", output: str, allow_failure: bool self, ctx: RepoContext, output: str, allow_failure: bool
) -> None: ) -> None:
# Prefer token path if indices unsupported (new nix) # Prefer token path if indices unsupported (new nix)
if self._indices_supported is False: if self._indices_supported is False:
@@ -215,7 +215,7 @@ class NixFlakeInstaller(BaseInstaller):
s = (stderr or "").lower() s = (stderr or "").lower()
return "no longer supports indices" in s or "does not support indices" in s return "no longer supports indices" in s or "does not support indices" in s
def _upgrade_index(self, ctx: "RepoContext", idx: int) -> bool: def _upgrade_index(self, ctx: RepoContext, idx: int) -> bool:
cmd = f"nix profile upgrade --refresh {idx}" cmd = f"nix profile upgrade --refresh {idx}"
res = self._runner.run(ctx, cmd, allow_failure=True) res = self._runner.run(ctx, cmd, allow_failure=True)
@@ -228,7 +228,7 @@ class NixFlakeInstaller(BaseInstaller):
return res.returncode == 0 return res.returncode == 0
def _remove_index(self, ctx: "RepoContext", idx: int) -> None: def _remove_index(self, ctx: RepoContext, idx: int) -> None:
res = self._runner.run(ctx, f"nix profile remove {idx}", allow_failure=True) res = self._runner.run(ctx, f"nix profile remove {idx}", allow_failure=True)
if self._stderr_says_indices_unsupported(getattr(res, "stderr", "")): if self._stderr_says_indices_unsupported(getattr(res, "stderr", "")):
@@ -237,7 +237,7 @@ class NixFlakeInstaller(BaseInstaller):
if self._indices_supported is None: if self._indices_supported is None:
self._indices_supported = True self._indices_supported = True
def _remove_tokens_for_output(self, ctx: "RepoContext", output: str) -> None: def _remove_tokens_for_output(self, ctx: RepoContext, output: str) -> None:
tokens = self._profile.find_remove_tokens_for_output(ctx, self._runner, output) tokens = self._profile.find_remove_tokens_for_output(ctx, self._runner, output)
if not tokens: if not tokens:
return return

View File

@@ -1,4 +1,4 @@
from .inspector import NixProfileInspector from .inspector import NixProfileInspector
from .models import NixProfileEntry from .models import NixProfileEntry
__all__ = ["NixProfileInspector", "NixProfileEntry"] __all__ = ["NixProfileEntry", "NixProfileInspector"]

View File

@@ -1,6 +1,6 @@
from __future__ import annotations from __future__ import annotations
from typing import Any, List, TYPE_CHECKING from typing import TYPE_CHECKING, Any
from .matcher import ( from .matcher import (
entry_matches_output, entry_matches_output,
@@ -29,7 +29,7 @@ class NixProfileInspector:
- find_remove_tokens_for_store_prefixes() - find_remove_tokens_for_store_prefixes()
""" """
def list_json(self, ctx: "RepoContext", runner: "CommandRunner") -> dict[str, Any]: def list_json(self, ctx: RepoContext, runner: CommandRunner) -> dict[str, Any]:
res = runner.run(ctx, "nix profile list --json", allow_failure=False) res = runner.run(ctx, "nix profile list --json", allow_failure=False)
raw = extract_stdout_text(res) raw = extract_stdout_text(res)
return parse_profile_list_json(raw) return parse_profile_list_json(raw)
@@ -40,14 +40,14 @@ class NixProfileInspector:
def find_installed_indices_for_output( def find_installed_indices_for_output(
self, self,
ctx: "RepoContext", ctx: RepoContext,
runner: "CommandRunner", runner: CommandRunner,
output: str, output: str,
) -> List[int]: ) -> list[int]:
data = self.list_json(ctx, runner) data = self.list_json(ctx, runner)
entries = normalize_elements(data) entries = normalize_elements(data)
hits: List[int] = [] hits: list[int] = []
for e in entries: for e in entries:
if e.index is None: if e.index is None:
continue continue
@@ -58,10 +58,10 @@ class NixProfileInspector:
def find_indices_by_store_path( def find_indices_by_store_path(
self, self,
ctx: "RepoContext", ctx: RepoContext,
runner: "CommandRunner", runner: CommandRunner,
store_path: str, store_path: str,
) -> List[int]: ) -> list[int]:
needle = (store_path or "").strip() needle = (store_path or "").strip()
if not needle: if not needle:
return [] return []
@@ -69,7 +69,7 @@ class NixProfileInspector:
data = self.list_json(ctx, runner) data = self.list_json(ctx, runner)
entries = normalize_elements(data) entries = normalize_elements(data)
hits: List[int] = [] hits: list[int] = []
for e in entries: for e in entries:
if e.index is None: if e.index is None:
continue continue
@@ -84,10 +84,10 @@ class NixProfileInspector:
def find_remove_tokens_for_output( def find_remove_tokens_for_output(
self, self,
ctx: "RepoContext", ctx: RepoContext,
runner: "CommandRunner", runner: CommandRunner,
output: str, output: str,
) -> List[str]: ) -> list[str]:
""" """
Returns profile remove tokens to remove entries matching a given output. Returns profile remove tokens to remove entries matching a given output.
@@ -101,7 +101,7 @@ class NixProfileInspector:
data = self.list_json(ctx, runner) data = self.list_json(ctx, runner)
entries = normalize_elements(data) entries = normalize_elements(data)
tokens: List[str] = [ tokens: list[str] = [
out out
] # critical: matches nix's own suggestion for conflicts ] # critical: matches nix's own suggestion for conflicts
@@ -119,7 +119,7 @@ class NixProfileInspector:
# stable unique preserving order # stable unique preserving order
seen: set[str] = set() seen: set[str] = set()
uniq: List[str] = [] uniq: list[str] = []
for t in tokens: for t in tokens:
if t and t not in seen: if t and t not in seen:
uniq.append(t) uniq.append(t)
@@ -128,10 +128,10 @@ class NixProfileInspector:
def find_remove_tokens_for_store_prefixes( def find_remove_tokens_for_store_prefixes(
self, self,
ctx: "RepoContext", ctx: RepoContext,
runner: "CommandRunner", runner: CommandRunner,
prefixes: List[str], prefixes: list[str],
) -> List[str]: ) -> list[str]:
""" """
Returns remove tokens for entries whose store path matches any prefix. Returns remove tokens for entries whose store path matches any prefix.
""" """
@@ -143,7 +143,7 @@ class NixProfileInspector:
data = self.list_json(ctx, runner) data = self.list_json(ctx, runner)
entries = normalize_elements(data) entries = normalize_elements(data)
tokens: List[str] = [] tokens: list[str] = []
for e in entries: for e in entries:
if not e.store_paths: if not e.store_paths:
continue continue
@@ -156,7 +156,7 @@ class NixProfileInspector:
tokens.append(n) tokens.append(n)
seen: set[str] = set() seen: set[str] = set()
uniq: List[str] = [] uniq: list[str] = []
for t in tokens: for t in tokens:
if t and t not in seen: if t and t not in seen:
uniq.append(t) uniq.append(t)

View File

@@ -1,7 +1,5 @@
from __future__ import annotations from __future__ import annotations
from typing import List
from .models import NixProfileEntry 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) 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() seen: set[int] = set()
uniq: List[int] = [] uniq: list[int] = []
for v in values: for v in values:
if v in seen: if v in seen:
continue continue

View File

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

View File

@@ -1,12 +1,13 @@
from __future__ import annotations from __future__ import annotations
import re import re
from typing import Any, Dict, Iterable, List, Optional from collections.abc import Iterable
from typing import Any
from .models import NixProfileEntry 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: Nix JSON schema varies:
- elements keys might be "0", "1", ... - elements keys might be "0", "1", ...
@@ -20,7 +21,7 @@ def coerce_index(key: str, entry: Dict[str, Any]) -> Optional[int]:
if k.isdigit(): if k.isdigit():
try: try:
return int(k) return int(k)
except Exception: except ValueError:
return None return None
# 2) Explicit index fields (schema-dependent) # 2) Explicit index fields (schema-dependent)
@@ -31,7 +32,7 @@ def coerce_index(key: str, entry: Dict[str, Any]) -> Optional[int]:
if isinstance(v, str) and v.strip().isdigit(): if isinstance(v, str) and v.strip().isdigit():
try: try:
return int(v.strip()) return int(v.strip())
except Exception: except ValueError:
pass pass
# 3) Last resort: extract trailing number from key if it looks like "<name>-<n>" # 3) Last resort: extract trailing number from key if it looks like "<name>-<n>"
@@ -39,13 +40,13 @@ def coerce_index(key: str, entry: Dict[str, Any]) -> Optional[int]:
if m: if m:
try: try:
return int(m.group(1)) return int(m.group(1))
except Exception: except ValueError:
return None return None
return None return None
def iter_store_paths(entry: Dict[str, Any]) -> Iterable[str]: def iter_store_paths(entry: dict[str, Any]) -> Iterable[str]:
""" """
Yield all possible store paths from a nix profile JSON entry. Yield all possible store paths from a nix profile JSON entry.
@@ -72,7 +73,7 @@ def iter_store_paths(entry: Dict[str, Any]) -> Iterable[str]:
outs = entry.get("outputs") outs = entry.get("outputs")
if isinstance(outs, dict): if isinstance(outs, dict):
for _, ov in outs.items(): for ov in outs.values():
if isinstance(ov, dict): if isinstance(ov, dict):
p = ov.get("storePath") p = ov.get("storePath")
if isinstance(p, str): if isinstance(p, str):
@@ -87,7 +88,7 @@ def normalize_store_path(store_path: str) -> str:
return (store_path or "").strip() 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. Converts nix profile list JSON into a list of normalized entries.
@@ -99,7 +100,7 @@ def normalize_elements(data: Dict[str, Any]) -> List[NixProfileEntry]:
if not isinstance(elements, dict): if not isinstance(elements, dict):
return [] return []
normalized: List[NixProfileEntry] = [] normalized: list[NixProfileEntry] = []
for k, entry in elements.items(): for k, entry in elements.items():
if not isinstance(entry, dict): if not isinstance(entry, dict):
@@ -109,7 +110,7 @@ def normalize_elements(data: Dict[str, Any]) -> List[NixProfileEntry]:
name = str(entry.get("name", "") or "") name = str(entry.get("name", "") or "")
attr = str(entry.get("attrPath", "") or "") attr = str(entry.get("attrPath", "") or "")
store_paths: List[str] = [] store_paths: list[str] = []
for p in iter_store_paths(entry): for p in iter_store_paths(entry):
sp = normalize_store_path(p) sp = normalize_store_path(p)
if sp: if sp:

View File

@@ -1,10 +1,10 @@
from __future__ import annotations from __future__ import annotations
import json 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`. Parse JSON output from `nix profile list --json`.

View File

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

View File

@@ -2,13 +2,15 @@ from __future__ import annotations
import random import random
import time import time
from collections.abc import Iterable
from dataclasses import dataclass from dataclasses import dataclass
from typing import Iterable, TYPE_CHECKING from typing import TYPE_CHECKING
from .types import RunResult from .types import RunResult
if TYPE_CHECKING: if TYPE_CHECKING:
from pkgmgr.actions.install.context import RepoContext from pkgmgr.actions.install.context import RepoContext
from .runner import CommandRunner from .runner import CommandRunner
@@ -31,8 +33,8 @@ class GitHubRateLimitRetry:
def run_with_retry( def run_with_retry(
self, self,
ctx: "RepoContext", ctx: RepoContext,
runner: "CommandRunner", runner: CommandRunner,
install_cmd: str, install_cmd: str,
) -> RunResult: ) -> RunResult:
quiet = bool(getattr(ctx, "quiet", False)) quiet = bool(getattr(ctx, "quiet", False))

View File

@@ -1,7 +1,6 @@
from __future__ import annotations from __future__ import annotations
import subprocess import subprocess
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from .types import RunResult from .types import RunResult
@@ -16,7 +15,7 @@ class CommandRunner:
Supports preview mode and compact failure output logging. Supports preview mode and compact failure output logging.
""" """
def run(self, ctx: "RepoContext", cmd: str, allow_failure: bool) -> RunResult: def run(self, ctx: RepoContext, cmd: str, allow_failure: bool) -> RunResult:
repo_dir = getattr(ctx, "repo_dir", None) or getattr(ctx, "repo_path", None) repo_dir = getattr(ctx, "repo_dir", None) or getattr(ctx, "repo_path", None)
preview = bool(getattr(ctx, "preview", False)) preview = bool(getattr(ctx, "preview", False))
quiet = bool(getattr(ctx, "quiet", False)) quiet = bool(getattr(ctx, "quiet", False))
@@ -32,8 +31,7 @@ class CommandRunner:
shell=True, shell=True,
cwd=repo_dir, cwd=repo_dir,
check=False, check=False,
stdout=subprocess.PIPE, capture_output=True,
stderr=subprocess.PIPE,
text=True, text=True,
) )
except Exception as e: except Exception as e:

View File

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

View File

@@ -36,7 +36,7 @@ class ArchPkgbuildInstaller(BaseInstaller):
try: try:
if hasattr(os, "geteuid") and os.geteuid() == 0: if hasattr(os, "geteuid") and os.geteuid() == 0:
return False return False
except Exception: except (AttributeError, OSError):
# On non-POSIX platforms just ignore this check. # On non-POSIX platforms just ignore this check.
pass pass

View File

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

View File

@@ -1,6 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" """
Installer for RPM-based packages defined in *.spec files. Installer for RPM-based packages defined in *.spec files.
@@ -14,12 +11,12 @@ This installer:
It targets RPM-based systems (Fedora / RHEL / CentOS / Rocky / Alma, etc.). It targets RPM-based systems (Fedora / RHEL / CentOS / Rocky / Alma, etc.).
""" """
from __future__ import annotations
import glob import glob
import os import os
import shutil import shutil
import tarfile import tarfile
from typing import List, Optional, Tuple
from pkgmgr.actions.install.context import RepoContext from pkgmgr.actions.install.context import RepoContext
from pkgmgr.actions.install.installers.base import BaseInstaller from pkgmgr.actions.install.installers.base import BaseInstaller
@@ -53,7 +50,7 @@ class RpmSpecInstaller(BaseInstaller):
return has_dnf or has_yum or has_yum_builddep 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.""" """Return the first *.spec file in the repository root, if any."""
pattern = os.path.join(ctx.repo_dir, "*.spec") pattern = os.path.join(ctx.repo_dir, "*.spec")
matches = sorted(glob.glob(pattern)) matches = sorted(glob.glob(pattern))
@@ -92,7 +89,7 @@ class RpmSpecInstaller(BaseInstaller):
for sub in ("BUILD", "BUILDROOT", "RPMS", "SOURCES", "SPECS", "SRPMS"): for sub in ("BUILD", "BUILDROOT", "RPMS", "SOURCES", "SPECS", "SRPMS"):
os.makedirs(os.path.join(topdir, sub), exist_ok=True) 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. Parse Name and Version from the given .spec file.
@@ -101,7 +98,7 @@ class RpmSpecInstaller(BaseInstaller):
name = None name = None
version = 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: for raw_line in f:
line = raw_line.strip() line = raw_line.strip()
# Ignore comments # Ignore comments
@@ -183,7 +180,7 @@ class RpmSpecInstaller(BaseInstaller):
return self._spec_path(ctx) is not None 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. Find RPMs built by rpmbuild.
@@ -202,9 +199,7 @@ class RpmSpecInstaller(BaseInstaller):
if shutil.which("dnf") is not None: if shutil.which("dnf") is not None:
cmd = f"sudo dnf builddep -y {spec_basename}" cmd = f"sudo dnf builddep -y {spec_basename}"
elif shutil.which("yum-builddep") is not None: elif shutil.which("yum-builddep") is not None or shutil.which("yum") is not None:
cmd = f"sudo yum-builddep -y {spec_basename}"
elif shutil.which("yum") is not None:
cmd = f"sudo yum-builddep -y {spec_basename}" cmd = f"sudo yum-builddep -y {spec_basename}"
else: else:
print( print(
@@ -215,7 +210,7 @@ class RpmSpecInstaller(BaseInstaller):
run_command(cmd, cwd=ctx.repo_dir, preview=ctx.preview) 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. Install or upgrade the built RPMs.

View File

@@ -4,8 +4,8 @@ from __future__ import annotations
import os import os
import sys import sys
from pkgmgr.actions.install.installers.base import BaseInstaller
from pkgmgr.actions.install.context import RepoContext from pkgmgr.actions.install.context import RepoContext
from pkgmgr.actions.install.installers.base import BaseInstaller
from pkgmgr.core.command.run import run_command from pkgmgr.core.command.run import run_command

View File

@@ -1,6 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" """
CLI layer model for the pkgmgr installation pipeline. CLI layer model for the pkgmgr installation pipeline.
@@ -19,7 +16,6 @@ from __future__ import annotations
import os import os
from enum import Enum from enum import Enum
from typing import Optional
class CliLayer(str, Enum): class CliLayer(str, Enum):
@@ -38,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. Return a numeric priority index for a given layer.
@@ -70,13 +66,11 @@ def classify_command_layer(command: str, repo_dir: str) -> CliLayer:
home = os.path.expanduser("~") home = os.path.expanduser("~")
# OS package managers # OS package managers
if command_abs.startswith("/usr/") or command_abs.startswith("/bin/"): if command_abs.startswith(("/usr/", "/bin/")):
return CliLayer.OS_PACKAGES return CliLayer.OS_PACKAGES
# Nix store / profile # Nix store / profile
if command_abs.startswith("/nix/store/") or command_abs.startswith( if command_abs.startswith(("/nix/store/", os.path.join(home, ".nix-profile"))):
os.path.join(home, ".nix-profile")
):
return CliLayer.NIX return CliLayer.NIX
# User-local bin # User-local bin

View File

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

View File

@@ -9,19 +9,20 @@ Public API:
""" """
from __future__ import annotations from __future__ import annotations
from .types import Repository, MirrorMap
from .list_cmd import list_mirrors
from .diff_cmd import diff_mirrors from .diff_cmd import diff_mirrors
from .list_cmd import list_mirrors
from .merge_cmd import merge_mirrors from .merge_cmd import merge_mirrors
from .setup_cmd import setup_mirrors from .setup_cmd import setup_mirrors
from .types import MirrorMap, Repository
from .visibility_cmd import set_mirror_visibility from .visibility_cmd import set_mirror_visibility
__all__ = [ __all__ = [
"Repository",
"MirrorMap", "MirrorMap",
"list_mirrors", "Repository",
"diff_mirrors", "diff_mirrors",
"list_mirrors",
"merge_mirrors", "merge_mirrors",
"setup_mirrors",
"set_mirror_visibility", "set_mirror_visibility",
"setup_mirrors",
] ]

View File

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

View File

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

View File

@@ -1,9 +1,7 @@
from __future__ import annotations from __future__ import annotations
import os import os
from typing import Optional, Set
from pkgmgr.core.git.errors import GitRunError
from pkgmgr.core.git.commands import ( from pkgmgr.core.git.commands import (
GitAddRemoteError, GitAddRemoteError,
GitAddRemotePushUrlError, GitAddRemotePushUrlError,
@@ -12,6 +10,7 @@ from pkgmgr.core.git.commands import (
add_remote_push_url, add_remote_push_url,
set_remote_url, set_remote_url,
) )
from pkgmgr.core.git.errors import GitRunError
from pkgmgr.core.git.queries import get_remote_push_urls, list_remotes from pkgmgr.core.git.queries import get_remote_push_urls, list_remotes
from .types import MirrorMap, RepoMirrorContext, Repository from .types import MirrorMap, RepoMirrorContext, Repository
@@ -38,13 +37,10 @@ def _is_git_remote_url(url: str) -> bool:
if u.startswith("ssh://"): if u.startswith("ssh://"):
return True return True
if (u.startswith("https://") or u.startswith("http://")) and u.endswith(".git"): return bool((u.startswith(("https://", "http://"))) and u.endswith(".git"))
return True
return False
def build_default_ssh_url(repo: Repository) -> Optional[str]: def build_default_ssh_url(repo: Repository) -> str | None:
provider = repo.get("provider") provider = repo.get("provider")
account = repo.get("account") account = repo.get("account")
name = repo.get("repository") name = repo.get("repository")
@@ -66,7 +62,7 @@ def _git_mirrors_only(m: MirrorMap) -> MirrorMap:
def determine_primary_remote_url( def determine_primary_remote_url(
repo: Repository, repo: Repository,
ctx: RepoMirrorContext, ctx: RepoMirrorContext,
) -> Optional[str]: ) -> str | None:
""" """
Priority order (GIT URLS ONLY): Priority order (GIT URLS ONLY):
1. origin from resolved mirrors (if it is a git URL) 1. origin from resolved mirrors (if it is a git URL)
@@ -80,7 +76,7 @@ def determine_primary_remote_url(
return origin return origin
for mirrors in (ctx.file_mirrors, ctx.config_mirrors): 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): if url and _is_git_remote_url(url):
return 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. Non-git URLs (like PyPI) are ignored and will never land in git config.
""" """
git_only = _git_mirrors_only(mirrors) 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: if not desired:
return return

View File

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

View File

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

View File

@@ -1,7 +1,6 @@
from __future__ import annotations from __future__ import annotations
import os import os
from typing import Dict, List, Tuple, Optional
import yaml import yaml
@@ -11,13 +10,12 @@ from .context import build_context
from .io import write_mirrors_file from .io import write_mirrors_file
from .types import MirrorMap, Repository from .types import MirrorMap, Repository
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
# Helpers # Helpers
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
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. Normalised key for identifying a repository in config files.
""" """
@@ -28,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. Load a user config YAML file as dict.
Non-dicts yield {}. Non-dicts yield {}.
@@ -37,10 +35,10 @@ def _load_user_config(path: str) -> Dict[str, object]:
return {} return {}
try: try:
with open(path, "r", encoding="utf-8") as f: with open(path, encoding="utf-8") as f:
data = yaml.safe_load(f) or {} data = yaml.safe_load(f) or {}
return data if isinstance(data, dict) else {} return data if isinstance(data, dict) else {}
except Exception: except (OSError, UnicodeDecodeError, yaml.YAMLError):
return {} return {}
@@ -50,13 +48,13 @@ def _load_user_config(path: str) -> Dict[str, object]:
def merge_mirrors( def merge_mirrors(
selected_repos: List[Repository], selected_repos: list[Repository],
repositories_base_dir: str, repositories_base_dir: str,
all_repos: List[Repository], all_repos: list[Repository],
source: str, source: str,
target: str, target: str,
preview: bool = False, preview: bool = False,
user_config_path: Optional[str] = None, user_config_path: str | None = None,
) -> None: ) -> None:
""" """
Merge mirrors between config and MIRRORS file. Merge mirrors between config and MIRRORS file.
@@ -74,8 +72,8 @@ def merge_mirrors(
""" """
# Load user config once if we intend to write to it. # Load user config once if we intend to write to it.
user_cfg: Optional[Dict[str, object]] = None user_cfg: dict[str, object] | None = None
user_cfg_path_expanded: Optional[str] = None user_cfg_path_expanded: str | None = None
if target == "config" and user_config_path and not preview: if target == "config" and user_config_path and not preview:
user_cfg_path_expanded = os.path.expanduser(user_config_path) user_cfg_path_expanded = os.path.expanduser(user_config_path)
@@ -130,7 +128,7 @@ def merge_mirrors(
repos = user_cfg.get("repositories") repos = user_cfg.get("repositories")
target_key = _repo_key(repo) target_key = _repo_key(repo)
existing_repo: Optional[Repository] = None existing_repo: Repository | None = None
# Find existing repo entry # Find existing repo entry
for entry in repos: for entry in repos:

View File

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

View File

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

View File

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

View File

@@ -2,10 +2,9 @@
from __future__ import annotations from __future__ import annotations
from urllib.parse import urlparse from urllib.parse import urlparse
from typing import Optional, Tuple
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() url = (url or "").strip()
if not url: if not url:
return "", None return "", None
@@ -58,7 +57,7 @@ def _strip_dot_git(name: str) -> str:
return n 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. Parse (host, owner, repo_name) from common Git remote URLs.

View File

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

View File

@@ -1,9 +1,10 @@
import os import os
from pkgmgr.core.repository.identifier import get_repo_identifier
from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.command.run import run_command
import sys import sys
from pkgmgr.core.command.run import run_command
from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier
def exec_proxy_command( def exec_proxy_command(
proxy_prefix: str, proxy_prefix: str,

View File

@@ -1,6 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" """
Backwards-compatible facade for the release file update helpers. Backwards-compatible facade for the release file update helpers.
@@ -13,23 +10,23 @@ Keep this package stable so existing imports continue to work, e.g.:
from __future__ import annotations from __future__ import annotations
from .editor import _open_editor_for_changelog
from .pyproject import update_pyproject_version
from .flake import update_flake_version
from .pkgbuild import update_pkgbuild_version
from .rpm_spec import update_spec_version
from .changelog_md import update_changelog from .changelog_md import update_changelog
from .debian import _get_debian_author, update_debian_changelog from .debian import _get_debian_author, update_debian_changelog
from .editor import _open_editor_for_changelog
from .flake import update_flake_version
from .pkgbuild import update_pkgbuild_version
from .pyproject import update_pyproject_version
from .rpm_changelog import update_spec_changelog from .rpm_changelog import update_spec_changelog
from .rpm_spec import update_spec_version
__all__ = [ __all__ = [
"_get_debian_author",
"_open_editor_for_changelog", "_open_editor_for_changelog",
"update_pyproject_version", "update_changelog",
"update_debian_changelog",
"update_flake_version", "update_flake_version",
"update_pkgbuild_version", "update_pkgbuild_version",
"update_spec_version", "update_pyproject_version",
"update_changelog",
"_get_debian_author",
"update_debian_changelog",
"update_spec_changelog", "update_spec_changelog",
"update_spec_version",
] ]

View File

@@ -0,0 +1,99 @@
from __future__ import annotations
import contextlib
import os
import re
import shutil
import subprocess
import tempfile
_HEADING = re.compile(r"^\s*#{1,6}\s+(.*?)\s*#*\s*$")
_INLINE_CODE = re.compile(r"`([^`\n]+)`")
_FINDING = re.compile(r"\bMD\d{3}\b")
_MULTI_BLANK = re.compile(r"\n{3,}")
class ChangelogLintError(RuntimeError):
"""Raised when a changelog entry cannot be made markdown-lint clean."""
def transform_changelog_message(text: str) -> str:
"""Normalise a free-form release message into the changelog house style.
A leading ``#`` heading becomes a bold line of its own (markdown
headings inside an entry body would collide with the ``## [version]``
structure), and inline ``code`` spans become ``*italic*`` so the entry
stays free of backticks.
"""
text = _INLINE_CODE.sub(r"*\1*", text)
out: list[str] = []
for line in text.split("\n"):
heading = _HEADING.match(line)
if heading:
if out and out[-1].strip():
out.append("")
out.append(f"**{heading.group(1).strip()}**")
out.append("")
else:
out.append(line)
joined = _MULTI_BLANK.sub("\n\n", "\n".join(out))
return joined.strip()
def lint_changelog_entry(reference_path: str, entry: str) -> list[str]:
"""Return markdown-lint findings for *entry* (empty list when clean).
The entry is checked inside a minimal ``# Changelog`` document so the
result reflects only the entry, not pre-existing issues in the target
file. ``markdownlint-cli2`` is used when available (picking up the
target repository's own config, i.e. the same rules the repo enforces);
otherwise a small built-in check runs.
"""
document = f"# Changelog\n\n{entry.strip()}\n"
if shutil.which("markdownlint-cli2"):
return _markdownlint(reference_path, document)
return _builtin_lint(document)
def _markdownlint(reference_path: str, document: str) -> list[str]:
directory = os.path.dirname(os.path.abspath(reference_path)) or "."
fd, tmp_path = tempfile.mkstemp(
suffix=".md", prefix=".pkgmgr-changelog-", dir=directory
)
name = os.path.basename(tmp_path)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(document)
proc = subprocess.run(
["markdownlint-cli2", name],
cwd=directory,
capture_output=True,
text=True,
check=False,
)
if proc.returncode == 0:
return []
output = f"{proc.stdout or ''}{proc.stderr or ''}"
findings = [
line.strip().replace(name, "changelog entry")
for line in output.splitlines()
if _FINDING.search(line)
]
return findings or [output.strip() or "markdown-lint reported an error"]
finally:
with contextlib.suppress(OSError):
os.remove(tmp_path)
def _builtin_lint(document: str) -> list[str]:
errors: list[str] = []
for number, line in enumerate(document.split("\n"), start=1):
if line != line.rstrip():
errors.append(f"line {number}: trailing whitespace (MD009)")
if "`" in line:
errors.append(f"line {number}: backtick is not allowed (use *italic*)")
if "\n\n\n" in document:
errors.append("multiple consecutive blank lines (MD012)")
return errors

View File

@@ -2,9 +2,14 @@ from __future__ import annotations
import os import os
import re import re
import sys
from datetime import date from datetime import date
from typing import Optional
from .changelog_lint import (
ChangelogLintError,
lint_changelog_entry,
transform_changelog_message,
)
from .editor import _open_editor_for_changelog from .editor import _open_editor_for_changelog
H1_RE = re.compile(r"^#\s+\S", re.MULTILINE) H1_RE = re.compile(r"^#\s+\S", re.MULTILINE)
@@ -49,7 +54,7 @@ def _insert_after_h1(existing: str, entry: str) -> str:
def update_changelog( def update_changelog(
changelog_path: str, changelog_path: str,
new_version: str, new_version: str,
message: Optional[str] = None, message: str | None = None,
preview: bool = False, preview: bool = False,
) -> str: ) -> str:
"""Insert a new release entry into CHANGELOG.md. """Insert a new release entry into CHANGELOG.md.
@@ -58,32 +63,51 @@ def update_changelog(
missing) and above any existing release entries, so the result stays missing) and above any existing release entries, so the result stays
markdown-lint-clean (MD041 first-line-h1, MD012 no-multiple-blanks). markdown-lint-clean (MD041 first-line-h1, MD012 no-multiple-blanks).
""" """
today = date.today().isoformat() today = date.today().isoformat() # noqa: DTZ011 - changelog dates are local calendar dates
if message is None: def _entry_for(raw: str) -> tuple[str, str]:
if preview: body = transform_changelog_message(raw).strip() or f"Release {new_version}"
message = "Automated release." return body, f"## [{new_version}] - {today}\n\n{body}\n\n"
else:
print( def _print_findings(findings: list[str]) -> None:
"\n[INFO] No release message provided, opening editor for changelog entry...\n" print("\n[ERROR] Changelog entry is not markdown-lint clean:")
for finding in findings:
print(f" - {finding}")
print()
if message is not None:
body, entry = _entry_for(message)
findings = lint_changelog_entry(changelog_path, entry)
if findings:
_print_findings(findings)
raise ChangelogLintError(
"Provided changelog message is not markdown-lint clean."
) )
editor_message = _open_editor_for_changelog() elif preview or not sys.stdin.isatty():
if not editor_message: body, entry = _entry_for(message or f"Release {new_version}")
message = "Automated release."
else: else:
message = editor_message attempt: str | None = None
while True:
entry = f"## [{new_version}] - {today}\n\n* {message}\n\n" print(
"\n[INFO] Provide the changelog entry — a leading '#' becomes "
"bold, `code` becomes italic.\n"
)
raw = _open_editor_for_changelog(attempt)
body, entry = _entry_for(raw or f"Release {new_version}")
findings = lint_changelog_entry(changelog_path, entry)
if not findings:
break
_print_findings(findings)
attempt = body
print("[INFO] Re-opening the editor so you can fix the entry...")
changelog = ""
if os.path.exists(changelog_path): if os.path.exists(changelog_path):
try: try:
with open(changelog_path, "r", encoding="utf-8") as f: with open(changelog_path, encoding="utf-8") as f:
changelog = f.read() changelog = f.read()
except Exception as exc: except (OSError, UnicodeDecodeError) as exc:
print(f"[WARN] Could not read existing CHANGELOG.md: {exc}") print(f"[WARN] Could not read existing CHANGELOG.md: {exc}")
changelog = ""
else:
changelog = ""
new_changelog = _insert_after_h1(changelog, entry) new_changelog = _insert_after_h1(changelog, entry)
@@ -93,10 +117,10 @@ def update_changelog(
if preview: if preview:
print(f"[PREVIEW] Would insert new entry for {new_version} into CHANGELOG.md") print(f"[PREVIEW] Would insert new entry for {new_version} into CHANGELOG.md")
return message return body
with open(changelog_path, "w", encoding="utf-8") as f: with open(changelog_path, "w", encoding="utf-8") as f:
f.write(new_changelog) f.write(new_changelog)
print(f"Updated CHANGELOG.md with version {new_version}") print(f"Updated CHANGELOG.md with version {new_version}")
return message return body

View File

@@ -2,12 +2,11 @@ from __future__ import annotations
import os import os
from datetime import datetime from datetime import datetime
from typing import Optional, Tuple
from pkgmgr.core.git.queries import get_config_value 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") name = os.environ.get("DEBFULLNAME")
email = os.environ.get("DEBEMAIL") email = os.environ.get("DEBEMAIL")
@@ -33,7 +32,7 @@ def update_debian_changelog(
debian_changelog_path: str, debian_changelog_path: str,
package_name: str, package_name: str,
new_version: str, new_version: str,
message: Optional[str] = None, message: str | None = None,
preview: bool = False, preview: bool = False,
) -> None: ) -> None:
if not os.path.exists(debian_changelog_path): if not os.path.exists(debian_changelog_path):
@@ -47,10 +46,17 @@ def update_debian_changelog(
author_name, author_email = _get_debian_author() author_name, author_email = _get_debian_author()
first_line = f"{package_name} ({debian_version}) unstable; urgency=medium" first_line = f"{package_name} ({debian_version}) unstable; urgency=medium"
body_line = message.strip() if message else f"Automated release {new_version}." body = (
message.strip()
if message and message.strip()
else f"Automated release {new_version}."
)
indented_body = "\n".join(
f" {line}" if line.strip() else line for line in body.split("\n")
)
stanza = ( stanza = (
f"{first_line}\n\n" f"{first_line}\n\n"
f" * {body_line}\n\n" f"{indented_body}\n\n"
f" -- {author_name} <{author_email}> {date_str}\n\n" f" -- {author_name} <{author_email}> {date_str}\n\n"
) )
@@ -62,9 +68,9 @@ def update_debian_changelog(
return return
try: try:
with open(debian_changelog_path, "r", encoding="utf-8") as f: with open(debian_changelog_path, encoding="utf-8") as f:
existing = f.read() existing = f.read()
except Exception as exc: except (OSError, UnicodeDecodeError) as exc:
print(f"[WARN] Could not read debian/changelog: {exc}") print(f"[WARN] Could not read debian/changelog: {exc}")
existing = "" existing = ""

View File

@@ -1,12 +1,12 @@
from __future__ import annotations from __future__ import annotations
import contextlib
import os import os
import subprocess import subprocess
import tempfile 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") editor = os.environ.get("EDITOR", "nano")
with tempfile.NamedTemporaryFile( with tempfile.NamedTemporaryFile(
@@ -16,9 +16,10 @@ def _open_editor_for_changelog(initial_message: Optional[str] = None) -> str:
) as tmp: ) as tmp:
tmp_path = tmp.name tmp_path = tmp.name
tmp.write( tmp.write(
"# Write the changelog entry for this release.\n" "; Write the changelog entry for this release.\n"
"# Lines starting with '#' will be ignored.\n" "; Lines starting with ';' are ignored.\n"
"# Empty result will fall back to a generic message.\n\n" "; A leading '#' becomes bold; `code` becomes italic.\n"
"; Empty result will fall back to a generic message.\n\n"
) )
if initial_message: if initial_message:
tmp.write(initial_message.strip() + "\n") tmp.write(initial_message.strip() + "\n")
@@ -33,13 +34,11 @@ def _open_editor_for_changelog(initial_message: Optional[str] = None) -> str:
) )
try: try:
with open(tmp_path, "r", encoding="utf-8") as f: with open(tmp_path, encoding="utf-8") as f:
content = f.read() content = f.read()
finally: finally:
try: with contextlib.suppress(OSError):
os.remove(tmp_path) os.remove(tmp_path)
except OSError:
pass
lines = [line for line in content.splitlines() if not line.strip().startswith("#")] lines = [line for line in content.splitlines() if not line.strip().startswith(";")]
return "\n".join(lines).strip() return "\n".join(lines).strip()

View File

@@ -12,9 +12,9 @@ def update_flake_version(
return return
try: try:
with open(flake_path, "r", encoding="utf-8") as f: with open(flake_path, encoding="utf-8") as f:
content = f.read() content = f.read()
except Exception as exc: except (OSError, UnicodeDecodeError) as exc:
print(f"[WARN] Could not read flake.nix: {exc}") print(f"[WARN] Could not read flake.nix: {exc}")
return return

View File

@@ -12,9 +12,9 @@ def update_pkgbuild_version(
return return
try: try:
with open(pkgbuild_path, "r", encoding="utf-8") as f: with open(pkgbuild_path, encoding="utf-8") as f:
content = f.read() content = f.read()
except Exception as exc: except (OSError, UnicodeDecodeError) as exc:
print(f"[WARN] Could not read PKGBUILD: {exc}") print(f"[WARN] Could not read PKGBUILD: {exc}")
return return

View File

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

View File

@@ -2,7 +2,6 @@ from __future__ import annotations
import os import os
from datetime import datetime from datetime import datetime
from typing import Optional
from .debian import _get_debian_author from .debian import _get_debian_author
@@ -11,7 +10,7 @@ def update_spec_changelog(
spec_path: str, spec_path: str,
package_name: str, package_name: str,
new_version: str, new_version: str,
message: Optional[str] = None, message: str | None = None,
preview: bool = False, preview: bool = False,
) -> None: ) -> None:
if not os.path.exists(spec_path): if not os.path.exists(spec_path):
@@ -19,9 +18,9 @@ def update_spec_changelog(
return return
try: try:
with open(spec_path, "r", encoding="utf-8") as f: with open(spec_path, encoding="utf-8") as f:
content = f.read() content = f.read()
except Exception as exc: except (OSError, UnicodeDecodeError) as exc:
print(f"[WARN] Could not read spec file for changelog update: {exc}") print(f"[WARN] Could not read spec file for changelog update: {exc}")
return return
@@ -30,11 +29,18 @@ def update_spec_changelog(
date_str = now.strftime("%a %b %d %Y") date_str = now.strftime("%a %b %d %Y")
author_name, author_email = _get_debian_author() author_name, author_email = _get_debian_author()
body_line = message.strip() if message else f"Automated release {new_version}." body = (
message.strip()
if message and message.strip()
else f"Automated release {new_version}."
)
dashed_body = "\n".join(
f"- {line}" if line.strip() else line for line in body.split("\n")
)
stanza = ( stanza = (
f"* {date_str} {author_name} <{author_email}> - {debian_version}\n" f"* {date_str} {author_name} <{author_email}> - {debian_version}\n"
f"- {body_line}\n\n" f"{dashed_body}\n\n"
) )
marker = "%changelog" marker = "%changelog"
@@ -57,7 +63,7 @@ def update_spec_changelog(
try: try:
with open(spec_path, "w", encoding="utf-8") as f: with open(spec_path, "w", encoding="utf-8") as f:
f.write(new_content) f.write(new_content)
except Exception as exc: except (OSError, UnicodeEncodeError) as exc:
print(f"[WARN] Failed to write updated spec changelog section: {exc}") print(f"[WARN] Failed to write updated spec changelog section: {exc}")
return return

View File

@@ -15,9 +15,9 @@ def update_spec_version(
return return
try: try:
with open(spec_path, "r", encoding="utf-8") as f: with open(spec_path, encoding="utf-8") as f:
content = f.read() content = f.read()
except Exception as exc: except (OSError, UnicodeDecodeError) as exc:
print(f"[WARN] Could not read spec file: {exc}") print(f"[WARN] Could not read spec file: {exc}")
return return

View File

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

View File

@@ -1,6 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" """
Version discovery and bumping helpers for the release workflow. Version discovery and bumping helpers for the release workflow.
""" """
@@ -10,10 +7,10 @@ from __future__ import annotations
from pkgmgr.core.git.queries import get_tags from pkgmgr.core.git.queries import get_tags
from pkgmgr.core.version.semver import ( from pkgmgr.core.version.semver import (
SemVer, SemVer,
find_latest_version,
bump_major, bump_major,
bump_minor, bump_minor,
bump_patch, bump_patch,
find_latest_version,
) )

View File

@@ -2,7 +2,6 @@ from __future__ import annotations
import os import os
import sys import sys
from typing import Optional
from pkgmgr.actions.branch import close_branch from pkgmgr.actions.branch import close_branch
from pkgmgr.core.git import GitRunError, run from pkgmgr.core.git import GitRunError, run
@@ -34,7 +33,7 @@ def _release_impl(
pyproject_path: str = "pyproject.toml", pyproject_path: str = "pyproject.toml",
changelog_path: str = "CHANGELOG.md", changelog_path: str = "CHANGELOG.md",
release_type: str = "patch", release_type: str = "patch",
message: Optional[str] = None, message: str | None = None,
preview: bool = False, preview: bool = False,
close: bool = False, close: bool = False,
force: bool = False, force: bool = False,
@@ -87,9 +86,8 @@ def _release_impl(
else: else:
print("[INFO] No RPM spec file found. Skipping spec version update.") print("[INFO] No RPM spec file found. Skipping spec version update.")
effective_message: Optional[str] = message effective_message: str | None = message
if effective_message is None and isinstance(changelog_message, str): if isinstance(changelog_message, str) and changelog_message.strip():
if changelog_message.strip():
effective_message = changelog_message.strip() effective_message = changelog_message.strip()
package_name = resolve_package_name(paths) package_name = resolve_package_name(paths)
@@ -188,7 +186,7 @@ def _release_impl(
print(f"[INFO] Deleting branch {branch} after successful release...") print(f"[INFO] Deleting branch {branch} after successful release...")
try: try:
close_branch(name=branch, base_branch="main", cwd=".") close_branch(name=branch, base_branch="main", cwd=".")
except Exception as exc: except (RuntimeError, GitRunError) as exc:
print(f"[WARN] Failed to close branch {branch} automatically: {exc}") print(f"[WARN] Failed to close branch {branch} automatically: {exc}")
@@ -196,7 +194,7 @@ def release(
pyproject_path: str = "pyproject.toml", pyproject_path: str = "pyproject.toml",
changelog_path: str = "CHANGELOG.md", changelog_path: str = "CHANGELOG.md",
release_type: str = "patch", release_type: str = "patch",
message: Optional[str] = None, message: str | None = None,
preview: bool = False, preview: bool = False,
force: bool = False, force: bool = False,
close: bool = False, close: bool = False,

View File

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

View File

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

View File

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

View File

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

View File

@@ -10,10 +10,12 @@ from pkgmgr.core.git.commands import (
push_upstream, push_upstream,
) )
DEFAULT_BRANCH = "main"
class GitBootstrapper: class GitBootstrapper:
def init_repo(self, repo_dir: str, preview: bool) -> None: def init_repo(self, repo_dir: str, preview: bool) -> None:
init(cwd=repo_dir, preview=preview) init(initial_branch=DEFAULT_BRANCH, cwd=repo_dir, preview=preview)
add_all(cwd=repo_dir, preview=preview) add_all(cwd=repo_dir, preview=preview)
try: try:
commit("Initial commit", cwd=repo_dir, preview=preview) commit("Initial commit", cwd=repo_dir, preview=preview)
@@ -21,15 +23,8 @@ class GitBootstrapper:
print(f"[WARN] Initial commit failed (continuing): {exc}") print(f"[WARN] Initial commit failed (continuing): {exc}")
def push_default_branch(self, repo_dir: str, preview: bool) -> None: def push_default_branch(self, repo_dir: str, preview: bool) -> None:
branch_move(DEFAULT_BRANCH, cwd=repo_dir, preview=preview)
try: try:
branch_move("main", cwd=repo_dir, preview=preview) push_upstream("origin", DEFAULT_BRANCH, cwd=repo_dir, preview=preview)
push_upstream("origin", "main", cwd=repo_dir, preview=preview)
return
except GitPushUpstreamError:
pass
try:
branch_move("master", cwd=repo_dir, preview=preview)
push_upstream("origin", "master", cwd=repo_dir, preview=preview)
except GitPushUpstreamError as exc: except GitPushUpstreamError as exc:
print(f"[WARN] Push failed: {exc}") print(f"[WARN] Push failed: {exc}")

View File

@@ -1,11 +1,11 @@
from __future__ import annotations 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.io import write_mirrors_file
from pkgmgr.actions.mirror.setup_cmd import setup_mirrors from pkgmgr.actions.mirror.setup_cmd import setup_mirrors
Repository = Dict[str, Any] Repository = dict[str, Any]
class MirrorBootstrapper: class MirrorBootstrapper:

View File

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

View File

@@ -1,7 +1,6 @@
from __future__ import annotations from __future__ import annotations
import re import re
from typing import Tuple
from urllib.parse import urlparse from urllib.parse import urlparse
from .model import RepoParts 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) 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: if ":" in host:
h, p = host.split(":", 1) h, p = host.split(":", 1)
return h, p or None return h, p or None
@@ -58,7 +57,7 @@ def _split_host_port(host: str) -> Tuple[str, str | None]:
def _strip_git_suffix(name: str) -> str: def _strip_git_suffix(name: str) -> str:
return name[:-4] if name.endswith(".git") else name return name.removesuffix(".git")
def _ensure_valid_repo_name(name: str) -> None: def _ensure_valid_repo_name(name: str) -> None:

View File

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

View File

@@ -1,23 +1,23 @@
from __future__ import annotations from __future__ import annotations
import os import os
from typing import Dict, Any from typing import Any
from pkgmgr.core.git.queries import get_config_value from pkgmgr.core.git.queries import get_config_value
from .parser import parse_identifier
from .planner import CreateRepoPlanner
from .config_writer import ConfigRepoWriter from .config_writer import ConfigRepoWriter
from .templates import TemplateRenderer
from .git_bootstrap import GitBootstrapper from .git_bootstrap import GitBootstrapper
from .mirrors import MirrorBootstrapper from .mirrors import MirrorBootstrapper
from .parser import parse_identifier
from .planner import CreateRepoPlanner
from .templates import TemplateRenderer
class CreateRepoService: class CreateRepoService:
def __init__( def __init__(
self, self,
*, *,
config_merged: Dict[str, Any], config_merged: dict[str, Any],
user_config_path: str, user_config_path: str,
bin_dir: str, bin_dir: str,
): ):

View File

@@ -2,13 +2,13 @@ from __future__ import annotations
import os import os
from pathlib import Path from pathlib import Path
from typing import Dict, Any from typing import Any
from pkgmgr.core.git.queries import get_repo_root from pkgmgr.core.git.queries import get_repo_root
try: try:
from jinja2 import Environment, FileSystemLoader, StrictUndefined from jinja2 import Environment, FileSystemLoader, StrictUndefined
except Exception as exc: # pragma: no cover except ImportError as exc: # pragma: no cover
Environment = None # type: ignore Environment = None # type: ignore
FileSystemLoader = None # type: ignore FileSystemLoader = None # type: ignore
StrictUndefined = None # type: ignore StrictUndefined = None # type: ignore
@@ -25,7 +25,7 @@ class TemplateRenderer:
self, self,
*, *,
repo_dir: str, repo_dir: str,
context: Dict[str, Any], context: dict[str, Any],
preview: bool, preview: bool,
) -> None: ) -> None:
if preview: if preview:

View File

@@ -1,7 +1,8 @@
import shutil
import os import os
from pkgmgr.core.repository.identifier import get_repo_identifier import shutil
from pkgmgr.core.repository.dir import get_repo_dir from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier
def delete_repos(selected_repos, repositories_base_dir, all_repos, preview=False): def delete_repos(selected_repos, repositories_base_dir, all_repos, preview=False):
@@ -27,7 +28,7 @@ def delete_repos(selected_repos, repositories_base_dir, all_repos, preview=False
print( print(
f"Deleted repository directory '{repo_dir}' for {repo_identifier}." f"Deleted repository directory '{repo_dir}' for {repo_identifier}."
) )
except Exception as e: except OSError as e:
print(f"Error deleting '{repo_dir}' for {repo_identifier}: {e}") print(f"Error deleting '{repo_dir}' for {repo_identifier}: {e}")
else: else:
print(f"Skipped deletion of '{repo_dir}' for {repo_identifier}.") print(f"Skipped deletion of '{repo_dir}' for {repo_identifier}.")

View File

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

View File

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

View File

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

View File

@@ -1,6 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations from __future__ import annotations
from pkgmgr.actions.update.manager import UpdateManager from pkgmgr.actions.update.manager import UpdateManager

View File

@@ -1,9 +1,7 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations from __future__ import annotations
from typing import Any, Iterable, List, Tuple from collections.abc import Iterable
from typing import Any
from pkgmgr.actions.update.system_updater import SystemUpdater 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.actions.repository.pull import pull_with_verification
from pkgmgr.core.repository.identifier import get_repo_identifier 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): for repo in list(selected_repos):
identifier = get_repo_identifier(repo, all_repos) identifier = get_repo_identifier(repo, all_repos)
@@ -59,7 +57,7 @@ class UpdateManager:
f"[Warning] update: pull failed for {identifier} (exit={code}). Continuing..." f"[Warning] update: pull failed for {identifier} (exit={code}). Continuing..."
) )
continue continue
except Exception as exc: except Exception as exc: # noqa: BLE001 - batch boundary: one repository must never abort the run
failures.append((identifier, f"pull failed: {exc}")) failures.append((identifier, f"pull failed: {exc}"))
if not quiet: if not quiet:
print( print(
@@ -90,7 +88,7 @@ class UpdateManager:
f"[Warning] update: install failed for {identifier} (exit={code}). Continuing..." f"[Warning] update: install failed for {identifier} (exit={code}). Continuing..."
) )
continue continue
except Exception as exc: except Exception as exc: # noqa: BLE001 - batch boundary: one repository must never abort the run
failures.append((identifier, f"install failed: {exc}")) failures.append((identifier, f"install failed: {exc}"))
if not quiet: if not quiet:
print( print(

View File

@@ -1,22 +1,18 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations from __future__ import annotations
import os import os
from dataclasses import dataclass 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. Parse /etc/os-release into a dict. Returns empty dict if missing.
""" """
if not os.path.exists(path): if not os.path.exists(path):
return {} return {}
result: Dict[str, str] = {} result: dict[str, str] = {}
with open(path, "r", encoding="utf-8") as f: with open(path, encoding="utf-8") as f:
for line in f: for line in f:
line = line.strip() line = line.strip()
if not line or line.startswith("#") or "=" not in line: if not line or line.startswith("#") or "=" not in line:
@@ -37,7 +33,7 @@ class OSReleaseInfo:
pretty_name: str = "" pretty_name: str = ""
@staticmethod @staticmethod
def load() -> "OSReleaseInfo": def load() -> OSReleaseInfo:
data = read_os_release() data = read_os_release()
return OSReleaseInfo( return OSReleaseInfo(
id=(data.get("ID") or "").lower(), id=(data.get("ID") or "").lower(),

View File

@@ -1,6 +1,3 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations from __future__ import annotations
import platform import platform

View File

@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
from __future__ import annotations from __future__ import annotations
import os import os
@@ -6,8 +5,8 @@ import os
from pkgmgr.core.config.load import load_config from pkgmgr.core.config.load import load_config
from .context import CLIContext from .context import CLIContext
from .parser import create_parser
from .dispatch import dispatch_command from .dispatch import dispatch_command
from .parser import create_parser
__all__ = ["CLIContext", "create_parser", "dispatch_command", "main"] __all__ = ["CLIContext", "create_parser", "dispatch_command", "main"]

View File

@@ -1,25 +1,27 @@
from .archive import handle_archive from .archive import handle_archive
from .repos import handle_repos_command
from .config import handle_config
from .tools import handle_tools_command
from .release import handle_release
from .publish import handle_publish
from .version import handle_version
from .make import handle_make
from .changelog import handle_changelog
from .branch import handle_branch from .branch import handle_branch
from .changelog import handle_changelog
from .code_scanning import handle_code_scanning
from .config import handle_config
from .make import handle_make
from .mirror import handle_mirror_command from .mirror import handle_mirror_command
from .publish import handle_publish
from .release import handle_release
from .repos import handle_repos_command
from .tools import handle_tools_command
from .version import handle_version
__all__ = [ __all__ = [
"handle_archive", "handle_archive",
"handle_repos_command",
"handle_config",
"handle_tools_command",
"handle_release",
"handle_publish",
"handle_version",
"handle_make",
"handle_changelog",
"handle_branch", "handle_branch",
"handle_changelog",
"handle_code_scanning",
"handle_config",
"handle_make",
"handle_mirror_command", "handle_mirror_command",
"handle_publish",
"handle_release",
"handle_repos_command",
"handle_tools_command",
"handle_version",
] ]

View File

@@ -2,8 +2,8 @@ from __future__ import annotations
import sys import sys
from pkgmgr.actions.branch import close_branch, drop_branch, open_branch
from pkgmgr.cli.context import CLIContext from pkgmgr.cli.context import CLIContext
from pkgmgr.actions.branch import open_branch, close_branch, drop_branch
def handle_branch(args, ctx: CLIContext) -> None: def handle_branch(args, ctx: CLIContext) -> None:

View File

@@ -2,23 +2,23 @@ from __future__ import annotations
import os import os
import sys import sys
from typing import Any, Dict, List, Optional, Tuple from typing import Any
from pkgmgr.actions.changelog import generate_changelog
from pkgmgr.cli.context import CLIContext from pkgmgr.cli.context import CLIContext
from pkgmgr.core.git import GitRunError
from pkgmgr.core.git.queries import get_tags
from pkgmgr.core.repository.dir import get_repo_dir from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier from pkgmgr.core.repository.identifier import get_repo_identifier
from pkgmgr.core.git.queries import get_tags
from pkgmgr.core.version.semver import extract_semver_from_tags from pkgmgr.core.version.semver import extract_semver_from_tags
from pkgmgr.actions.changelog import generate_changelog
Repository = dict[str, Any]
Repository = Dict[str, Any]
def _find_previous_and_current_tag( def _find_previous_and_current_tag(
tags: List[str], tags: list[str],
target_tag: Optional[str] = None, target_tag: str | None = None,
) -> Tuple[Optional[str], Optional[str]]: ) -> tuple[str | None, str | None]:
""" """
Given a list of tags and an optional target tag, determine Given a list of tags and an optional target tag, determine
(previous_tag, current_tag) on the SemVer axis. (previous_tag, current_tag) on the SemVer axis.
@@ -65,7 +65,7 @@ def _find_previous_and_current_tag(
def handle_changelog( def handle_changelog(
args, args,
ctx: CLIContext, ctx: CLIContext,
selected: List[Repository], selected: list[Repository],
) -> None: ) -> None:
""" """
Handle the 'changelog' command. Handle the 'changelog' command.
@@ -94,7 +94,7 @@ def handle_changelog(
if not repo_dir: if not repo_dir:
try: try:
repo_dir = get_repo_dir(ctx.repositories_base_dir, repo) repo_dir = get_repo_dir(ctx.repositories_base_dir, repo)
except Exception: except (AttributeError, KeyError, TypeError):
repo_dir = None repo_dir = None
identifier = get_repo_identifier(repo, ctx.all_repositories) identifier = get_repo_identifier(repo, ctx.all_repositories)
@@ -114,12 +114,12 @@ def handle_changelog(
try: try:
tags = get_tags(cwd=repo_dir) tags = get_tags(cwd=repo_dir)
except Exception as exc: except GitRunError as exc:
print(f"[ERROR] Could not read git tags: {exc}") print(f"[ERROR] Could not read git tags: {exc}")
tags = [] tags = []
from_ref: Optional[str] = None from_ref: str | None = None
to_ref: Optional[str] = None to_ref: str | None = None
if range_arg: if range_arg:
# Explicit range provided # Explicit range provided

View File

@@ -0,0 +1,24 @@
from __future__ import annotations
import sys
from pkgmgr.actions.code_scanning import CodeScanningError, download_code_scanning
from pkgmgr.cli.context import CLIContext
def handle_code_scanning(args, ctx: CLIContext) -> None:
try:
result = download_code_scanning(
repo=getattr(args, "repo", None),
output_dir=getattr(args, "output", None),
state=getattr(args, "state", None),
)
except CodeScanningError as exc:
print(f"[ERROR] {exc}", file=sys.stderr)
sys.exit(1)
print(f"[code-scanning] Repository: {result.repo}")
print(f"[code-scanning] Alerts: {result.alert_count}")
print(f"[code-scanning] Output: {result.output_dir}")
for path in result.files:
print(f" - {path}")

View File

@@ -1,27 +1,25 @@
# src/pkgmgr/cli/commands/config.py # src/pkgmgr/cli/commands/config.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations from __future__ import annotations
import os import os
import sys
import shutil import shutil
import sys
from pathlib import Path from pathlib import Path
from typing import Any, Dict, Optional from typing import Any
import yaml import yaml
from pkgmgr.cli.context import CLIContext
from pkgmgr.actions.config.init import config_init
from pkgmgr.actions.config.add import interactive_add from pkgmgr.actions.config.add import interactive_add
from pkgmgr.core.repository.resolve import resolve_repos from pkgmgr.actions.config.init import config_init
from pkgmgr.core.config.save import save_user_config
from pkgmgr.actions.config.show import show_config from pkgmgr.actions.config.show import show_config
from pkgmgr.cli.context import CLIContext
from pkgmgr.core.command.run import run_command from pkgmgr.core.command.run import run_command
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 Load the user config from ~/.config/pkgmgr/config.yaml
(or whatever ctx.user_config_path is), creating the directory if needed. (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) os.makedirs(cfg_dir, exist_ok=True)
if os.path.exists(user_config_path_expanded): 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 yaml.safe_load(f) or {"repositories": []}
return {"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 Find the directory inside the installed pkgmgr package that contains
the default config files. the default config files.
@@ -75,7 +73,7 @@ def _update_default_configs(user_config_path: str) -> None:
for name in os.listdir(source_dir): for name in os.listdir(source_dir):
lower = name.lower() lower = name.lower()
if not (lower.endswith(".yml") or lower.endswith(".yaml")): if not (lower.endswith((".yml", ".yaml"))):
continue continue
if name == "config.yaml": if name == "config.yaml":
continue continue

View File

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

View File

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

View File

@@ -1,17 +1,17 @@
from __future__ import annotations from __future__ import annotations
import os import os
from typing import Any, Dict, List from typing import Any
from pkgmgr.actions.publish import publish from pkgmgr.actions.publish import publish
from pkgmgr.cli.context import CLIContext from pkgmgr.cli.context import CLIContext
from pkgmgr.core.repository.dir import get_repo_dir from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier 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: if not selected:
print("[pkgmgr] No repositories selected for publish.") print("[pkgmgr] No repositories selected for publish.")
return return

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