DB backends (db)

The file-DB provider registry and its built-in backends. A third-party backend subclasses DbProvider with its own NAME and advertises it in the pkgforge.dbs entry-point group; see Adding a backend. The checks every backend must pass are in Contract suite.

pkgforge.exceptions.DbError

Bases: PkgForgeError, ValueError

A file DB's on-disk content could not be used as one.

Raised for a parse error, non-UTF-8 bytes, content that is not in the chosen format, a JSON Lines line or YAML top-level document that is not a mapping, or a record that breaks the schema (see :func:_check_record). Caught by :func:pkgforge.cli.main's error boundary like any :class:~pkgforge.exceptions.PkgForgeError (one stderr line, exit 1). A :class:ValueError because it reports malformed input.

pkgforge.db.Db = typing.Dict[str, 'typing.Optional[FileEntry]'] module-attribute

pkgforge.db.DbProvider(path)

Bases: Registered, ABC

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

A subclass registers by declaring its own NAME (plus, optionally, ALIASES); see :class:pkgforge._registry.Registered. SUFFIXES (file suffixes that infer this format from a --db path -- case- insensitive; a missing leading dot is added, since :attr:Path.suffix always includes one and a dotless entry could otherwise never match; a suffix with more than one dot, e.g. ".tar.gz", raises ValueError at class-creation time, since :attr:Path.suffix only ever returns the last dot-segment and such an entry could never match either) are normalized -- and validated -- from the class's own body before registration, so a bad suffix leaves the class entirely unregistered rather than half-registered.

Source code in src/pkgforge/db/__init__.py

def __init__(self, path: Path) -> None:
    self.path = path

ALIASES = () class-attribute

NAME = '' class-attribute

SUFFIXES = () class-attribute

path = path instance-attribute

__init_subclass__(**kwargs)

Source code in src/pkgforge/db/__init__.py

def __init_subclass__(cls, **kwargs: typing.Any) -> None:
    # Validate this class's own suffixes before Registered registers its
    # NAME/ALIASES, so a bad suffix never leaves a name half-registered.
    if "SUFFIXES" in cls.__dict__:
        normalized = []
        for suffix in cls.__dict__["SUFFIXES"]:
            suffix = suffix.lower()
            if suffix and not suffix.startswith("."):
                suffix = "." + suffix
            if suffix.count(".") > 1:
                raise ValueError(
                    f"suffix {suffix!r} has more than one dot; Path.suffix "
                    "only ever returns the last dot-segment, so this could "
                    "never match a --db path"
                )
            normalized.append(suffix)
        cls.SUFFIXES = tuple(normalized)
    super().__init_subclass__(**kwargs)

add(path, entry) abstractmethod

Record entry for path.

Source code in src/pkgforge/db/__init__.py

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

batch()

Optionally batch a run of add/remove calls for efficiency.

A context manager yielding self. The default implementation (this one) does nothing extra -- every add/remove inside it still writes (and, for the append-log backends, still locks) exactly as it would outside one -- so an existing or third-party provider that doesn't override this keeps working unchanged. :class:~pkgforge.db.SQLiteDb overrides it to hold one connection open across the whole batch and commit periodically instead of connecting, creating the schema and committing once per call.

Source code in src/pkgforge/db/__init__.py

@contextlib.contextmanager
def batch(self) -> typing.Iterator["DbProvider"]:
    """Optionally batch a run of ``add``/``remove`` calls for efficiency.

    A context manager yielding ``self``. The default implementation
    (this one) does nothing extra -- every ``add``/``remove`` inside it
    still writes (and, for the append-log backends, still locks)
    exactly as it would outside one -- so an existing or third-party
    provider that doesn't override this keeps working unchanged.
    :class:`~pkgforge.db.SQLiteDb` overrides it to hold one
    connection open across the whole batch and commit periodically
    instead of connecting, creating the schema and committing once per
    call.
    """
    yield self

compact() abstractmethod

Collapse redundant history: keep every live entry and drop the tombstones of removed paths, so load() holds no None after it.

Source code in src/pkgforge/db/__init__.py

@abc.abstractmethod
def compact(self) -> None:
    """Collapse redundant history: keep every live entry and drop the
    tombstones of removed paths, so ``load()`` holds no ``None`` after it."""

init() abstractmethod

Create or reset an empty DB.

Source code in src/pkgforge/db/__init__.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/__init__.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/__init__.py

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

pkgforge.db.open_db(path, fmt=None)

Resolve and construct the :class:DbProvider for path.

An explicit fmt wins; otherwise the suffix decides, defaulting to JSON Lines. The file's content is never inspected: a file that is not in the resolved format fails when it is loaded.

Source code in src/pkgforge/db/__init__.py

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

    An explicit ``fmt`` wins; otherwise the suffix decides, defaulting to JSON
    Lines. The file's content is never inspected: a file that is not in the
    resolved format fails when it is loaded.
    """
    return DbProvider.get(fmt or format_for_suffix(path))(path)

pkgforge.db.format_for_suffix(path)

Infer a format from path's suffix, else :data:DEFAULT_FORMAT.

A suffix no registered class holds loads the advertised pkgforge.dbs entry points (a plugin that fails to load is logged and skipped) and looks again.

Source code in src/pkgforge/db/__init__.py

def format_for_suffix(path: Path) -> str:
    """Infer a format from ``path``'s suffix, else :data:`DEFAULT_FORMAT`.

    A suffix no registered class holds loads the advertised ``pkgforge.dbs``
    entry points (a plugin that fails to load is logged and skipped) and looks
    again.
    """
    suffix = path.suffix.lower()
    for attempt in range(2):
        for registered in set(DbProvider._registry.values()):
            # Own class dict only: a subclass that does not redeclare SUFFIXES
            # itself must not silently claim its parent's (plain attribute
            # access would find the inherited value instead).
            if suffix in registered.__dict__.get("SUFFIXES", ()):
                return registered.NAME
        if attempt == 0:
            DbProvider._load_pending(_LOGGER)
    return DEFAULT_FORMAT

pkgforge.db.jsonl_line(path, entry)

One JSON Lines record for path/entry (None writes a removal marker), terminated with a single newline: the exact text :class:JSONLinesDb appends.

Source code in src/pkgforge/db/_jsonl.py

def jsonl_line(path: str, entry: typing.Optional["FileEntry"]) -> str:
    """One JSON Lines record for ``path``/``entry`` (``None`` writes a removal
    marker), terminated with a single newline: the exact text
    :class:`JSONLinesDb` appends."""
    return json.dumps({"path": path, **_fields(entry)}, sort_keys=True) + "\n"

pkgforge.db.parse_jsonl(text, source='text')

Parse JSON Lines text (what :func:jsonl_line writes) into a :data:~pkgforge.db.Db; the last record for a path wins.

source names the text in errors, which read {source}:{line}: .... Raises :class:~pkgforge.exceptions.DbError for a malformed record.

Source code in src/pkgforge/db/_jsonl.py

def parse_jsonl(text: str, source: str = "text") -> Db:
    """Parse JSON Lines ``text`` (what :func:`jsonl_line` writes) into a
    :data:`~pkgforge.db.Db`; the last record for a path wins.

    ``source`` names the text in errors, which read ``{source}:{line}: ...``.
    Raises :class:`~pkgforge.exceptions.DbError` for a malformed record.
    """
    return _parse_jsonl(text, source, "")

Built-in backends

pkgforge.db.JSONLinesDb(path)

Bases: AppendLogDb

Append-only JSON Lines: one JSON object per line, last per path wins.

Source code in src/pkgforge/db/__init__.py

def __init__(self, path: Path) -> None:
    self.path = path

NAME = 'jsonl' class-attribute instance-attribute

SUFFIXES = ('.jsonl', '.ndjson') class-attribute instance-attribute

load()

Source code in src/pkgforge/db/_jsonl.py

def load(self) -> Db:
    if not self.path.exists():
        return {}
    hint = _format_hint(self.NAME)
    return _parse_jsonl(_read_text(self.path, hint), self.path, hint)

pkgforge.db.YAMLDb(path)

Bases: AppendLogDb

Append-only YAML: a single mapping, appended key by key -- the last duplicate key wins.

Reading relies on the YAML loader letting a later duplicate mapping key win -- a property of this backend's format, not a general guarantee. A null value is a removal.

Source code in src/pkgforge/db/__init__.py

def __init__(self, path: Path) -> None:
    self.path = path

NAME = 'yaml' class-attribute instance-attribute

SUFFIXES = ('.yaml', '.yml') class-attribute instance-attribute

init()

Source code in src/pkgforge/db/_yaml.py

def init(self) -> None:
    # An empty file needs no parser, but a DB this install cannot read
    # back should not be created.
    _yaml_io()
    super().init()

load()

Source code in src/pkgforge/db/_yaml.py

def load(self) -> Db:
    if not self.path.exists():
        return {}
    loader, _ = _yaml_io()
    import yaml

    hint = _format_hint(self.NAME)
    try:
        text = self.path.read_text(encoding="utf-8-sig")
        if _too_deep(text):
            raise DbError(f"{self.path}: nested too deeply")
        data = yaml.load(text, Loader=loader)
    except UnicodeDecodeError as exc:
        raise DbError(f"{self.path}: {exc}{hint}") from exc
    except yaml.YAMLError as exc:
        detail = " ".join(str(exc).split())  # PyYAML reports over several lines
        raise DbError(f"{self.path}: {detail}{hint}") from exc
    except RecursionError as exc:
        raise DbError(f"{self.path}: nested too deeply") from exc
    if data is None:
        return {}
    if not isinstance(data, dict):
        raise DbError(
            f"{self.path}: invalid YAML file DB: top-level document is "
            f"not a mapping{hint}"
        )
    return {path: _check_record(path, rec, self.path) for path, rec in data.items()}

pkgforge.db.SQLiteDb(path)

Bases: DbProvider

SQLite store: one row per path, upserted in place (no append log).

Source code in src/pkgforge/db/__init__.py

def __init__(self, path: Path) -> None:
    self.path = path

NAME = 'sqlite' class-attribute instance-attribute

SUFFIXES = ('.db', '.sqlite', '.sqlite3') class-attribute instance-attribute

add(path, entry)

Source code in src/pkgforge/db/_sqlite.py

def add(self, path: str, entry: "FileEntry") -> None:
    if self._batch_conn is not None:
        self._insert_add(self._batch_conn, path, entry)
        self._batch_commit_if_due()
        return
    with self._connect() as conn:
        self._insert_add(conn, path, entry)

batch()

Hold one connection open across many add/remove calls, committing every :data:_BATCH_ROWS rows and once more on exit (exception included, so a killed batch keeps whatever it already committed, like the append-log backends) instead of connecting, creating the schema and committing once per call.

Not reentrant (a nested with provider.batch(): reuses the same connection and row counter, which is harmless but pointless); callers only ever open one at a time (see Project._batch).

Source code in src/pkgforge/db/_sqlite.py

@contextlib.contextmanager
def batch(self) -> typing.Iterator["SQLiteDb"]:
    """Hold one connection open across many ``add``/``remove`` calls,
    committing every :data:`_BATCH_ROWS` rows and once more on exit
    (exception included, so a killed batch keeps whatever it already
    committed, like the append-log backends) instead of connecting,
    creating the schema and committing once per call.

    Not reentrant (a nested ``with provider.batch():`` reuses the same
    connection and row counter, which is harmless but pointless);
    callers only ever open one at a time (see ``Project._batch``).
    """
    sqlite3 = _sqlite3()
    try:
        conn = sqlite3.connect(os.fspath(self.path))
        try:
            conn.execute(self._SCHEMA)
        except BaseException:
            conn.close()
            raise
        self._batch_conn = conn
        self._batch_count = 0
        try:
            yield self
        finally:
            try:
                conn.commit()
            finally:
                self._batch_conn = None
                conn.close()
    except sqlite3.Error as exc:
        raise _db_error(self.path, exc) from exc

compact()

Source code in src/pkgforge/db/_sqlite.py

def compact(self) -> None:
    if not self.path.exists():
        return
    with self._connect(ensure_schema=False) as conn:
        if not self._has_entries_table(conn):
            return
        conn.execute("DELETE FROM entries WHERE removed=1")
        # VACUUM must run outside a transaction; commit what we have first.
        conn.commit()
        conn.execute("VACUUM")

init()

Source code in src/pkgforge/db/_sqlite.py

def init(self) -> None:
    with self._connect(ensure_schema=False) as conn:
        conn.execute("DROP TABLE IF EXISTS entries")
        conn.execute(self._SCHEMA)

load()

Source code in src/pkgforge/db/_sqlite.py

def load(self) -> Db:
    if not self.path.exists():
        return {}
    db: Db = {}
    # ensure_schema=False: a read must never write, not even the no-op
    # `CREATE TABLE IF NOT EXISTS` -- pointing a read at an empty or
    # foreign SQLite file must not add a table to it.
    with self._connect(ensure_schema=False) as conn:
        if not self._has_entries_table(conn):
            return {}
        for row in conn.execute(
            'SELECT path, mode, owner, "group", type, kind, caps, package, '
            "meta_json, removed FROM entries"
        ):
            (
                path,
                mode,
                owner,
                group,
                type_,
                kind,
                caps,
                package,
                meta_json,
                removed,
            ) = row
            if removed:
                db[path] = _check_record(path, None, self.path)
                continue
            # pkgforge writes an empty meta as the literal "{}"; skip
            # json.loads for it and give each row its own fresh dict.
            if meta_json == "{}":
                meta = {}
            elif meta_json is None:
                meta = None
            else:
                try:
                    meta = json.loads(meta_json)
                except (TypeError, ValueError, RecursionError) as exc:
                    raise DbError(
                        f"{self.path}: {path!r}: invalid meta_json: {exc}"
                    ) from exc
            rec = {
                "mode": mode,
                "owner": owner,
                "group": group,
                "type": type_,
                "kind": kind,
                "caps": caps,
                "package": package,
                "meta": meta,
            }
            db[path] = (
                rec
                if _record_ok(path, rec)
                else _check_record(path, rec, self.path)
            )
    return db

remove(path)

Source code in src/pkgforge/db/_sqlite.py

def remove(self, path: str) -> None:
    if self._batch_conn is not None:
        self._insert_remove(self._batch_conn, path)
        self._batch_commit_if_due()
        return
    with self._connect() as conn:
        self._insert_remove(conn, path)