API Reference

Generated from docstrings. pkgforge is primarily a CLI, but its core types and dump helpers are importable.

Core types

pkgforge.common.FileType

Bases: str, Enum

Directory = 'directory' class-attribute instance-attribute

File = 'file' class-attribute instance-attribute

Symlink = 'symlink' class-attribute instance-attribute

from_path(path) classmethod

Source code in src/pkgforge/common.py

@classmethod
def from_path(cls, path: Path) -> "FileType":
    if path.is_symlink():
        return FileType.Symlink
    elif path.is_dir():
        return cls.Directory
    elif path.is_file():
        return cls.File
    else:
        raise TypeError(path)

pkgforge.common.FileEntry

Bases: TypedDict

group instance-attribute

meta instance-attribute

mode instance-attribute

owner instance-attribute

type instance-attribute

apply(path, chown=False, *, logger=None, usedefault=DEFAULT)

Source code in src/pkgforge/common.py

def apply(
    self,
    path: Path,
    chown=False,
    *,
    logger: logging.Logger = None,
    usedefault=DEFAULT,
):
    mode = self["mode"]
    owner = self["owner"]
    group = self["group"]

    if mode and mode != usedefault:
        mode = int(mode, 8) if isinstance(mode, str) else mode
        if logger:
            logger.debug("Setting mode for %s to %o", path, mode)
        os.chmod(path, mode, follow_symlinks=False)

    if chown and (owner != usedefault or group != usedefault):
        if pwd is None or grp is None:
            raise RuntimeError("chown requires the Unix pwd/grp modules")
        owner = -1 if owner == usedefault else pwd.getpwnam(owner).pw_uid
        group = -1 if group == usedefault else grp.getgrnam(group).gr_gid
        if logger:
            logger.debug("Setting owner/group for %s to %s:%s", path, owner, group)
        os.chown(path, owner, group, follow_symlinks=False)

from_args(args, **overwrite) classmethod

Source code in src/pkgforge/common.py

@classmethod
def from_args(cls, args: "FileEntryArgs", **overwrite) -> "FileEntry":
    entry = {
        "mode": args.mode,
        "owner": args.owner,
        "group": args.group,
        "type": FileType(args.type) if args.type else args.type,
        "meta": dict(args.meta),
    }
    entry.update(overwrite)
    return entry

from_path(path, meta=None) classmethod

Source code in src/pkgforge/common.py

@classmethod
def from_path(cls, path: Path, meta: "typing.Dict[str, str]" = None) -> "FileEntry":
    stat = path.lstat()
    owner = group = DEFAULT
    if pwd is not None:
        try:
            owner = pwd.getpwuid(stat.st_uid).pw_name
        except KeyError:
            owner = DEFAULT
    if grp is not None:
        try:
            group = grp.getgrgid(stat.st_gid).gr_name
        except KeyError:
            group = DEFAULT

    return {
        # Store mode as an octal permission string so the DB round-trips and
        # dumps (e.g. %attr(644,...)) are correct; apply() reads it back via
        # int(mode, 8).
        "mode": mode_to_octal(stat.st_mode),
        "owner": owner,
        "group": group,
        "type": FileType.from_path(path),
        "meta": {} if meta is None else meta,
    }

resolve_for(path, lookupval=AUTO, **overwrite)

Source code in src/pkgforge/common.py

def resolve_for(self, path: Path, lookupval=AUTO, **overwrite) -> "FileEntry":
    resolved: FileEntry = {**self}
    ondisk = FileEntry.from_path(path)
    for k, v in resolved.items():
        if v == lookupval:
            resolved[k] = ondisk[k]
    resolved.update(overwrite)

    return resolved

pkgforge.common.FileEntryArgs

Bases: Cmd

group = DEFAULT class-attribute instance-attribute

meta = {} class-attribute instance-attribute

mode = DEFAULT class-attribute instance-attribute

owner = DEFAULT class-attribute instance-attribute

type = None class-attribute instance-attribute

pkgforge.common.PkgForgeCmd

Bases: LoggingArgs, Cmd

Common base for every pkgforge subcommand.

Carries the two app-wide options (--db and --buildroot), the file-DB read/write helpers, and the build-root <-> local-path translation. Each leaf command subclasses this and implements __call__; a leaf attaches itself to the :class:PkgForge root's subcommand tree via :meth:_register.

buildroot = Path(buildroot) if buildroot else Path('.') class-attribute instance-attribute

db = Path(filedb) if filedb else None class-attribute instance-attribute

db_format = os.environ.get('PKGFORGE_DB_FORMAT') class-attribute instance-attribute

__call__()

Source code in src/pkgforge/common.py

def __call__(self):
    raise NotImplementedError(self)

add_entry(buildpath, entry)

Source code in src/pkgforge/common.py

def add_entry(self, buildpath: Path, entry: "FileEntry"):
    self._write_entry(buildpath, entry)

buildpath(localpath)

Source code in src/pkgforge/common.py

def buildpath(self, localpath: Path) -> Path:
    return Path("/", localpath.relative_to(self.buildroot))

compactdb()

Collapse the DB's redundant history (backend-specific; no-op if none).

Source code in src/pkgforge/common.py

def compactdb(self) -> None:
    """Collapse the DB's redundant history (backend-specific; no-op if none)."""
    if self._no_file_db():
        return
    self._provider(for_read=True).compact()

initdb()

Create or reset an empty DB (no-op for a stdout / unset DB).

Source code in src/pkgforge/common.py

def initdb(self) -> None:
    """Create or reset an empty DB (no-op for a stdout / unset DB)."""
    if self._no_file_db():
        return
    self._provider().init()

loaddb()

Source code in src/pkgforge/common.py

def loaddb(self) -> "typing.Dict[str, typing.Optional[FileEntry]]":
    if self._no_file_db():
        return {}
    return self._provider(for_read=True).load()

localpath(buildpath)

Source code in src/pkgforge/common.py

def localpath(self, buildpath: Path) -> Path:
    return Path(self.buildroot, *buildpath.parts[1:])

remove_entry(buildpath)

Source code in src/pkgforge/common.py

def remove_entry(self, buildpath: Path):
    self._write_entry(buildpath, None)

Storage backends

pkgforge.db.DbProvider(path)

Bases: ABC

Storage backend for a file DB, bound to a filesystem path.

Source code in src/pkgforge/db.py

def __init__(self, path: Path):
    self.path = path

format = '' class-attribute instance-attribute

path = path instance-attribute

add(path, entry) abstractmethod

Record entry for path.

Source code in src/pkgforge/db.py

@abc.abstractmethod
def add(self, path: str, entry: "FileEntry") -> None:
    """Record ``entry`` for ``path``."""

compact() abstractmethod

Collapse redundant history (a no-op for backends without any).

Source code in src/pkgforge/db.py

@abc.abstractmethod
def compact(self) -> None:
    """Collapse redundant history (a no-op for backends without any)."""

init() abstractmethod

Create or reset an empty DB.

Source code in src/pkgforge/db.py

@abc.abstractmethod
def init(self) -> None:
    """Create or reset an empty DB."""

load() abstractmethod

Return the full DB as {path: entry-or-None}.

Source code in src/pkgforge/db.py

@abc.abstractmethod
def load(self) -> "Db":
    """Return the full DB as ``{path: entry-or-None}``."""

remove(path) abstractmethod

Mark path removed.

Source code in src/pkgforge/db.py

@abc.abstractmethod
def remove(self, path: str) -> None:
    """Mark ``path`` removed."""

pkgforge.db.register_provider(name, provider_cls, *, suffixes=(), sniff=None)

Register a file-DB backend so :func:open_db can select it.

This is the extension seam that keeps third-party backends OUT of core pkgforge: an external package calls register_provider at import time and its format becomes usable via --db-format <name>, a matching file suffix, or content sniffing.

• name -- the format name (used by --db-format and error messages).
• provider_cls -- a :class:DbProvider subclass constructed as provider_cls(path).
• suffixes -- file suffixes (e.g. (".toml",), leading dot, case-insensitive) that infer this format from a --db path.
• sniff -- optional sniff(head: bytes) -> bool that inspects a file's first 16 bytes and returns True if this backend owns it. Registered sniffers are consulted newest-first, before the built-in heuristics, so a later registration can claim a shape an earlier one would.

Returns provider_cls so it can be used as a decorator. Re-registering a name replaces the previous class for that name.

Source code in src/pkgforge/db.py

def register_provider(
    name: str,
    provider_cls: "typing.Type[DbProvider]",
    *,
    suffixes: "typing.Iterable[str]" = (),
    sniff: "typing.Optional[Sniffer]" = None,
) -> "typing.Type[DbProvider]":
    """Register a file-DB backend so :func:`open_db` can select it.

    This is the extension seam that keeps third-party backends OUT of core
    pkgforge: an external package calls ``register_provider`` at import time
    and its format becomes usable via ``--db-format <name>``, a matching file
    suffix, or content sniffing.

    * ``name`` -- the format name (used by ``--db-format`` and error messages).
    * ``provider_cls`` -- a :class:`DbProvider` subclass constructed as
      ``provider_cls(path)``.
    * ``suffixes`` -- file suffixes (e.g. ``(".toml",)``, leading dot,
      case-insensitive) that infer this format from a ``--db`` path.
    * ``sniff`` -- optional ``sniff(head: bytes) -> bool`` that inspects a file's
      first 16 bytes and returns True if this backend owns it. Registered
      sniffers are consulted newest-first, before the built-in heuristics, so a
      later registration can claim a shape an earlier one would.

    Returns ``provider_cls`` so it can be used as a decorator. Re-registering a
    name replaces the previous class for that name.
    """
    PROVIDERS[name] = provider_cls
    for suffix in suffixes:
        SUFFIX_FORMATS[suffix.lower()] = name
    if sniff is not None:
        _SNIFFERS.insert(0, (name, sniff))
    return provider_cls

pkgforge.db.open_db(path, fmt=None, *, for_read=False)

Resolve and construct the :class:DbProvider for path.

Precedence: an explicit fmt wins; otherwise, when for_read and the file already exists, its content is sniffed (so a mislabeled or legacy file still loads); otherwise the suffix decides (defaulting to JSON Lines).

Source code in src/pkgforge/db.py

def open_db(
    path: Path, fmt: "typing.Optional[str]" = None, *, for_read: bool = False
) -> DbProvider:
    """Resolve and construct the :class:`DbProvider` for ``path``.

    Precedence: an explicit ``fmt`` wins; otherwise, when ``for_read`` and the
    file already exists, its content is sniffed (so a mislabeled or legacy file
    still loads); otherwise the suffix decides (defaulting to JSON Lines).
    """
    if fmt is None:
        detected = sniff_format(path) if (for_read and path.exists()) else None
        fmt = detected or format_for_suffix(path)
    try:
        provider_cls = PROVIDERS[fmt]
    except KeyError:
        raise ValueError(
            f"unknown db format {fmt!r}; choose from {', '.join(sorted(PROVIDERS))}"
        )
    return provider_cls(path)

Commands

pkgforge.install.Install(**kwargs)

Bases: FileEntryArgs, PkgForgeCmd

Install a source into the build root and record its file entry.

Source code in src/pkgforge/install.py

def __init__(self, **kwargs):
    if kwargs.pop("D", False):
        kwargs["no_target_directory"] = True
        kwargs["parents"] = True
    if kwargs.pop("d", False):
        kwargs["type"] = "directory"

    decompress = kwargs.get("decompress")
    if decompress is None or decompress == "-":
        kwargs["decompress"] = True
    super().__init__(**kwargs)

chown = False class-attribute instance-attribute

decompress = False class-attribute instance-attribute

destination instance-attribute

exclude = [] class-attribute instance-attribute

no_target_directory instance-attribute

noentry = False class-attribute instance-attribute

parents instance-attribute

remove_source = False class-attribute instance-attribute

source = [] class-attribute instance-attribute

type = DEFAULT class-attribute instance-attribute

__call__()

Source code in src/pkgforge/install.py

def __call__(self):
    if isinstance(self.source, list):
        for source in self.source:
            cloned = dict(self._get_kwargs())
            cloned["source"] = source
            Install(**cloned)()
        return

    if self.type == DEFAULT:
        self._logger_.debug("Determining type from source")
        if self.source and self.source != DEFAULT:
            self.type = FileType.from_path(self.source)
        else:
            self.type = FileType.File
    else:
        self.type = FileType(self.type)

    if self.decompress is True:
        self.decompress = self.source.suffix[1:]

    if not self.no_target_directory:
        if str(self.source) == DEFAULT:
            raise ValueError(self.source)
        self.destination = self.destination / self.source.name
        if self.decompress:
            self.destination = self.destination.with_name(
                self.destination.name.removesuffix(f".{self.decompress}")
            )
        if self.type == FileType.Directory:
            name = self.destination.name
            parts = name.split(".")
            if len(parts) > 1:
                suffixes = parts[1:]
                suffixes.reverse()
                for ty in ["tar", "iso"]:
                    if ty in suffixes:
                        idx = suffixes.index(ty)
                        name = ".".join([parts[0], *reversed(suffixes[idx + 1 :])])
                        self.destination = self.destination.with_name(name)
                        break

    if not self.destination.is_absolute() and not self.buildroot:
        raise ValueError(self.destination)

    dest = self.destination
    if self.buildroot:
        if dest.is_absolute():
            dest = Path(self.buildroot, *dest.parts[1:])
        else:
            dest = self.buildroot / dest

    if self.parents:
        dest.parent.mkdir(parents=True, exist_ok=True)

    self.install(self.source, dest)

    if self.remove_source and self.source not in [DEFAULT, None]:
        # A directory source needs rmtree; unlink only removes files/symlinks.
        if self.source.is_dir() and not self.source.is_symlink():
            shutil.rmtree(self.source)
        else:
            self.source.unlink()

    fileentry = FileEntry.from_args(self)
    fileentry = FileEntry.resolve_for(fileentry, dest)
    FileEntry.apply(fileentry, dest, chown=self.chown, logger=self._logger_)

    if not self.noentry:
        fspath = os.fspath(self.buildpath(dest))
        self.add_entry(fspath, fileentry)

install(src, dst)

Source code in src/pkgforge/install.py

def install(self, src: Path, dst: Path):
    self._logger_.info(
        "Installing %s at %s", DEFAULT if src is None else src, self.buildpath(dst)
    )
    if dst.exists() and src not in [DEFAULT, None] and src.resolve() == dst.resolve():
        return
    if self.type == FileType.File:
        dst.unlink(True)
        if not src:
            dst.touch()
        elif self.decompress:
            with dst.open("wb") as f:
                subprocess.run(
                    [
                        DECOMPRESS_CMDS.get(self.decompress, self.decompress),
                        "-kc",
                        os.fspath(src),
                    ],
                    stdin=sys.stdin.fileno(),
                    stdout=f,
                    check=True,
                )
        elif str(src) != DEFAULT:
            # Hardlink src -> dst. os.link works on every supported Python
            # (Path.link_to was removed in 3.12; Path.hardlink_to only
            # exists from 3.10), so use the stdlib os call directly.
            os.link(os.fspath(src), os.fspath(dst))
            shutil.copystat(src, dst, follow_symlinks=False)
        else:
            self._logger_.info("Obtaining data from stdin")
            with dst.open("wb") as output:
                if not sys.stdin.isatty():
                    with os.fdopen(sys.stdin.fileno(), "rb") as input:
                        shutil.copyfileobj(input, output)
    elif self.type == FileType.Symlink:
        if str(src) == DEFAULT or not src:
            target = self.meta.get("target")
            if not target:
                raise ValueError(src)
            dst.symlink_to(target)
        else:
            target = src.readlink()
            self.meta["target"] = os.fspath(target)
            dst.symlink_to(target)
            shutil.copystat(src, dst, follow_symlinks=False)

    elif self.type == FileType.Directory:
        dst.mkdir(exist_ok=True)
        if not src:
            ...
        elif src == DEFAULT or not src.is_dir():
            # Extract an archive source. Prefer stdlib tarfile for the tar
            # family (no external binary, cross-platform, safe `data`
            # filter); fall back to bsdtar for stdin and formats tarfile
            # can't open (e.g. iso).
            if src != DEFAULT and _is_tar_source(src):
                self._logger_.debug("Extracting %s via tarfile", src)
                _extract_tar(src, dst)
            else:
                self._logger_.debug("Extracting %s via bsdtar", src)
                subprocess.run(
                    [
                        "bsdtar",
                        "-x",
                        "-C",
                        os.fspath(dst),
                        "-f",
                        os.fspath(src) if src != DEFAULT else "-",
                    ],
                    stdin=sys.stdin.fileno() if src == DEFAULT else None,
                    check=True,
                )
        elif src.is_dir():
            shutil.copystat(src, dst, follow_symlinks=False)
            if self.exclude:
                filter = PathMatch(self.exclude, src)

                def _ignore(_dir: str, _files: "typing.List[str]"):
                    return [
                        file for file in _files if filter.match(Path(_dir, file))
                    ]

            else:
                _ignore = None
            shutil.copytree(
                src,
                dst,
                symlinks=True,
                ignore=_ignore,
                ignore_dangling_symlinks=True,
                dirs_exist_ok=True,
            )

    else:
        raise NotImplementedError(self.type)

pkgforge.scan.ScanCmd

Bases: FileEntryArgs, PkgForgeCmd

Scan a path under the build root and record each file's entry in the DB.

exclude = [] class-attribute instance-attribute

missing = False class-attribute instance-attribute

path instance-attribute

__call__()

Source code in src/pkgforge/scan.py

def __call__(self):
    db = self.loaddb() if self.missing else {}
    baseentry = FileEntry.from_args(self, type="--")
    scanpath = self.buildroot / self.path.lstrip("/")
    filter = PathMatch(self.exclude, scanpath)
    self._logger_.info("Scanning %s", scanpath)

    def _scanfile(path: Path):
        if self.exclude and filter.match(path):
            self._logger_.debug("Excluding %s", path)
            return
        fspath = os.fspath(self.buildpath(path))
        if db.get(fspath) is None:
            self._logger_.info("Updating file entry for: %s", fspath)
            self.add_entry(fspath, entry=FileEntry.resolve_for(baseentry, path))

    if not scanpath.is_dir():
        _scanfile(scanpath)
    else:
        for top, dirs, files in os.walk(scanpath):
            for file in [*dirs, *files]:
                _scanfile(Path(top, file))

pkgforge.compact.Compact

Bases: PkgForgeCmd

Collapse the DB to one record per live path (drop superseded/removed).

__call__()

Source code in src/pkgforge/compact.py

def __call__(self):
    if self.db is None or str(self.db) == "-":
        self._logger_.info("No file DB to compact")
        return
    before = self.loaddb()
    live = sum(1 for entry in before.values() if entry is not None)
    self.compactdb()
    self._logger_.info(
        "Compacted %s: %d live path(s), %d removal(s) dropped",
        self.db,
        live,
        len(before) - live,
    )

pkgforge.dbdump.DbDump

Bases: PkgForgeCmd

Dump the file DB into a packaging manifest (rpm or debian).

exclude = [] class-attribute instance-attribute

format instance-attribute

output = Path('-') class-attribute instance-attribute

__call__()

Source code in src/pkgforge/dbdump.py

def __call__(self):
    entries = self._surviving_entries()

    if self.format in PER_ENTRY_FORMATS:
        dumper = PER_ENTRY_FORMATS[self.format]
        out = None
        try:
            if str(self.output) == "-":
                out = os.fdopen(sys.stdout.fileno(), "wb", closefd=False)
            else:
                out = self.output.open("wb")
            for path, entry in entries:
                out.write(dumper(path, entry))
        finally:
            if out:
                out.flush()
                if str(self.output) != "-":
                    out.close()
        return

    if self.format in MULTI_ARTIFACT_FORMATS:
        artifacts = MULTI_ARTIFACT_FORMATS[self.format](entries)
        if str(self.output) == "-":
            out = os.fdopen(sys.stdout.fileno(), "wb", closefd=False)
            try:
                for name, data in artifacts.items():
                    out.write(f"# === {name} ===\n".encode())
                    out.write(data)
                out.flush()
            finally:
                pass  # do not close the shared stdout fd
        else:
            self.output.mkdir(parents=True, exist_ok=True)
            for name, data in artifacts.items():
                (self.output / name).write_bytes(data)
                self._logger_.info("Wrote %s", self.output / name)
        return

    raise SystemExit(
        f"unknown format {self.format!r}; choose from {', '.join(dump_formats())}"
    )

Dump formats

pkgforge.dbdump.rpmspecfile(path, entry)

Source code in src/pkgforge/dbdump.py

def rpmspecfile(path: str, entry: FileEntry) -> bytes:
    prefix = entry["meta"].get("rpmprefix") or ""
    if prefix:
        prefix += " "
    if entry["type"] == "directory":
        prefix += "%dir "

    return (
        f"{prefix}%attr({entry['mode']},{entry['owner']},{entry['group']}) "
        f"{json.dumps(path)}\n"
    ).encode()

pkgforge.dbdump.dump_formats()

All known format names, sorted (for --help / error messages).

Source code in src/pkgforge/dbdump.py

def dump_formats() -> "typing.List[str]":
    """All known format names, sorted (for --help / error messages)."""
    return sorted([*PER_ENTRY_FORMATS, *MULTI_ARTIFACT_FORMATS])

Extraction helpers

pkgforge.install._is_tar_source(src)

True if src is a tar-family archive stdlib :mod:tarfile can extract.

Source code in src/pkgforge/install.py

def _is_tar_source(src: "Path | str") -> bool:
    """True if ``src`` is a tar-family archive stdlib :mod:`tarfile` can extract."""
    name = os.fspath(src).lower()
    return name.endswith(TAR_SUFFIXES)

pkgforge.install._extract_tar(fileobj_or_name, dst)

Extract a tar-family archive into dst using stdlib :mod:tarfile.

Uses the safe data extraction filter where the interpreter supports it (guards against absolute paths / traversal / special files); older interpreters without filter= extract without it.

Source code in src/pkgforge/install.py

def _extract_tar(fileobj_or_name, dst: Path) -> None:
    """Extract a tar-family archive into ``dst`` using stdlib :mod:`tarfile`.

    Uses the safe ``data`` extraction filter where the interpreter supports it
    (guards against absolute paths / traversal / special files); older
    interpreters without ``filter=`` extract without it.
    """
    kwargs = {}
    if isinstance(fileobj_or_name, (str, os.PathLike)):
        opener = tarfile.open(name=os.fspath(fileobj_or_name), mode="r:*")
    else:
        opener = tarfile.open(fileobj=fileobj_or_name, mode="r|*")
    with opener as tar:
        if _TARFILE_HAS_FILTER:
            kwargs["filter"] = "data"
        tar.extractall(os.fspath(dst), **kwargs)