feat: unified package model with idempotent upgrades across all package managers

pyinfra-metadata.toml

@@ -39,18 +39,6 @@ path = "src/pyinfra/facts/brew.py"
 type = "fact"
 tags = ["package-manager", "system"]
 
-[pyinfra.plugins."choco-ops"]
-name = "choco"
-path = "src/pyinfra/operations/choco.py"
-type = "operation"
-tags = ["package-manager", "system"]
-
-[pyinfra.plugins."choco-facts"]
-name = "choco"
-path = "src/pyinfra/facts/choco.py"
-type = "fact"
-tags = ["package-manager", "system"]
-
 [pyinfra.plugins."dnf-ops"]
 name = "dnf"
 path = "src/pyinfra/operations/dnf.py"

src/pyinfra/facts/apk.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import re
+
 from typing_extensions import override
 
 from pyinfra.api import FactBase
@@ -10,6 +12,12 @@
 # Modified to return version and release inside a single group and removed extra capturing groups
 APK_REGEX = r"(.+)-([^-]+-r[^-]+) \S+ \{\S+\} \(.+?\)"
 
+# Regex for `apk version -v -l '<'` output lines like:
+#   vim-9.0.1-r0 < 9.0.2-r0
+# Package names can contain hyphens, so the version starts after the last
+# hyphen that is immediately followed by a digit.
+APK_UPGRADEABLE_REGEX = re.compile(r"^(.+)-(\d\S*)\s+<\s+(\S+)$")
+
 
 class ApkPackages(FactBase):
     """
@@ -35,3 +43,67 @@ def requires_command(self) -> str:
     @override
     def process(self, output):
         return parse_packages(APK_REGEX, output)
+
+
+class ApkUpgradeablePackages(FactBase):
+    """
+    Returns a dict of packages with available upgrades:
+
+    .. code:: python
+
+        {
+            "package_name": "available_version",
+        }
+    """
+
+    @override
+    def command(self) -> str:
+        return "apk version -v -l '<'"
+
+    @override
+    def requires_command(self) -> str:
+        return "apk"
+
+    default = dict
+    use_default_on_error = True
+
+    @override
+    def process(self, output):
+        upgradeable: dict[str, str] = {}
+        for line in output:
+            match = APK_UPGRADEABLE_REGEX.match(line)
+            if match:
+                name = match.group(1)
+                available_version = match.group(3)
+                upgradeable[name] = available_version
+        return upgradeable
+
+
+class ApkHeldPackages(FactBase):
+    """
+    Returns a list of held (locked) apk packages.
+
+    .. code:: python
+
+        ["package_name", ...]
+    """
+
+    @override
+    def command(self) -> str:
+        return "apk lock --list 2>/dev/null || true"
+
+    @override
+    def requires_command(self) -> str:
+        return "apk"
+
+    default = list
+    use_default_on_error = True
+
+    @override
+    def process(self, output):
+        held: list[str] = []
+        for line in output:
+            line = line.strip()
+            if line:
+                held.append(line)
+        return held

src/pyinfra/facts/brew.py

@@ -111,6 +111,101 @@ def requires_command(self) -> str:
         return "brew"
 
 
+BREW_OUTDATED_REGEX = re.compile(r"^(\S+)\s+\(([^)]+)\)\s+[<!=]+\s+(\S+)")
+
+
+class BrewOutdatedPackages(FactBase):
+    """
+    Returns a dict of outdated brew packages and their available versions:
+
+    .. code:: python
+
+        {
+            "package_name": "available_version",
+        }
+    """
+
+    use_default_on_error = True
+
+    @override
+    def command(self) -> str:
+        return "brew outdated --verbose 2>/dev/null || true"
+
+    @override
+    def requires_command(self) -> str:
+        return "brew"
+
+    default = dict
+
+    @override
+    def process(self, output):
+        outdated: dict[str, str] = {}
+        for line in output:
+            match = BREW_OUTDATED_REGEX.match(line)
+            if match:
+                outdated[match.group(1)] = match.group(3)
+        return outdated
+
+
+class BrewOutdatedCasks(FactBase):
+    """
+    Returns a dict of outdated brew casks and their available versions:
+
+    .. code:: python
+
+        {
+            "cask_name": "available_version",
+        }
+    """
+
+    use_default_on_error = True
+
+    @override
+    def command(self) -> str:
+        return "brew outdated --cask --verbose 2>/dev/null || true"
+
+    @override
+    def requires_command(self) -> str:
+        return "brew"
+
+    default = dict
+
+    @override
+    def process(self, output):
+        outdated: dict[str, str] = {}
+        for line in output:
+            match = BREW_OUTDATED_REGEX.match(line)
+            if match:
+                outdated[match.group(1)] = match.group(3)
+        return outdated
+
+
+class BrewPinnedPackages(FactBase):
+    """
+    Returns a list of pinned brew packages.
+
+    .. code:: python
+
+        ["package_name", ...]
+    """
+
+    use_default_on_error = True
+
+    @override
+    def command(self) -> str:
+        return "brew list --pinned 2>/dev/null || true"
+
+    @override
+    def requires_command(self) -> str:
+        return "brew"
+
+    default = list
+
+    @override
+    def process(self, output):
+        return [line.strip() for line in output if line.strip()]
+
+
 class BrewTaps(FactBase):
     """
     Returns a list of brew taps.

src/pyinfra/facts/deb.py

@@ -58,6 +58,80 @@ def process(self, output):
         return parse_packages(self.regex, output)
 
 
+class DebUpgradeablePackages(FactBase):
+    """
+    Returns a dict of upgradeable apt packages and their available versions:
+
+    .. code:: python
+
+        {
+            "package_name": "available_version",
+        }
+    """
+
+    @override
+    def command(self) -> str:
+        return "apt list --upgradeable -qq 2>/dev/null || true"
+
+    @override
+    def requires_command(self) -> str:
+        return "apt"
+
+    default = dict
+    use_default_on_error = True
+
+    @override
+    def process(self, output):
+        packages: dict[str, str] = {}
+
+        for line in output:
+            line = line.strip()
+            if not line:
+                continue
+            # Format: package_name/suite version arch [upgradable from: old_version]
+            match = re.match(r"^([^/]+)/\S+\s+(\S+)", line)
+            if match:
+                packages[match.group(1)] = match.group(2)
+
+        return packages
+
+
+class DebHeldPackages(FactBase):
+    """
+    Returns a list of held dpkg package names:
+
+    .. code:: python
+
+        ["package_name", ...]
+    """
+
+    @override
+    def command(self) -> str:
+        return "dpkg --get-selections | grep 'hold$' || true"
+
+    @override
+    def requires_command(self) -> str:
+        return "dpkg"
+
+    default = list
+    use_default_on_error = True
+
+    @override
+    def process(self, output):
+        packages: list[str] = []
+
+        for line in output:
+            line = line.strip()
+            if not line:
+                continue
+            # Format: package_name\thold
+            parts = line.split()
+            if parts:
+                packages.append(parts[0])
+
+        return packages
+
+
 class DebPackage(FactBase):
     """
     Returns information on a .deb archive or installed package.