viat.providers.tracker ¶

class GitFileTracker(TrackerBaseMixin, ViatFileTracker):
    """The git file tracker.

    Args:
        config: All configuration required for the tracker.
        resolver: A path resolver used to process incoming paths.
        static_config: Static configuration for the vault.
    """

    config: GitFileTrackerConfig
    """The configuration used to initialize the tracker."""

    resolver: ViatPathResolver | None
    """The resolver used to initialize the storage."""

    static_config: ViatVaultStaticConfig
    """The vault's static configuration."""

    _repo: pygit2.Repository

    def __init__(
        self,
        config: GitFileTrackerConfig,
        resolver: ViatPathResolver | None = None,
        static_config: ViatVaultStaticConfig | None = None,
    ) -> None:
        try:
            self._repo = pygit2.Repository(config.repo_root.as_posix())
        except pygit2.GitError as err:
            raise ViatFileTrackerError('The tracker root is not a git repository') from err

        self.config = config
        self.resolver = resolver
        self.static_config = static_config or ViatVaultStaticConfig()

    def _recurse_into_tree(self, tree: pygit2.Tree, base_path: pathlib.Path) -> Iterable[pathlib.Path]:
        for obj in tree:
            match obj:
                case pygit2.Blob():
                    if obj.name:
                        yield base_path / obj.name

                case pygit2.Tree():
                    if obj.name:
                        yield from self._recurse_into_tree(obj, base_path / obj.name)

    @override
    def iter_paths(self) -> Iterable[pathlib.Path]:
        try:
            ref = self._repo.revparse_single(self.config.revision)
        except KeyError:
            raise ViatFileTrackerError('No git HEAD pointer') from None

        if not isinstance(ref, pygit2.Commit):
            raise ViatFileTrackerError('git HEAD does not point to a commit')

        yield from self._recurse_into_tree(ref.tree, self.config.repo_root)

    @override
    def is_tracked(self, path: pathlib.Path | str) -> bool:
        rel_path = self._resolve_path(path)

        try:
            self._repo.blame(rel_path.as_posix())
        except (KeyError, ValueError):
            return False

        return True

@dataclass
class GitFileTrackerConfig:
    """Configuration for the git file tracker."""

    repo_root: pathlib.Path
    """The root of the repository to track files from."""

    revision: str = 'HEAD'
    """The git revision to track files from."""

class GlobFileTracker(TrackerBaseMixin, ViatFileTracker):
    """The default glob file tracker.

    Due to inconsistencies between [glob][] and [fnmatch][] from the standard library,
    we use the more flexible [wcmatch] library.

    [wcmatch]: https://facelessuser.github.io/wcmatch

    Args:
        config: All configuration required for the tracker.
        resolver: A path resolver used to process incoming paths.
        static_config: Static configuration for the vault.
    """

    config: GlobFileTrackerConfig
    """The configuration used to initialize the tracker."""

    resolver: ViatPathResolver | None
    """The resolver used to initialize the storage."""

    static_config: ViatVaultStaticConfig
    """The vault's static configuration."""

    def __init__(
        self,
        config: GlobFileTrackerConfig,
        resolver: ViatPathResolver | None = None,
        static_config: ViatVaultStaticConfig | None = None,
    ) -> None:
        self.config = config
        self.resolver = resolver
        self.static_config = static_config or ViatVaultStaticConfig()

    def _glob_flags(self) -> int:
        return sum(getattr(glob, f) for f in self.config.flags)

    @override
    def iter_paths(self) -> Iterable[pathlib.Path]:
        for raw_path in glob.glob(self.config.patterns, root_dir=self.config.root, flags=self._glob_flags()):
            yield pathlib.Path(raw_path)

    @override
    def is_tracked(self, path: pathlib.Path | str) -> bool:
        rel_path = self._resolve_path(path)
        return glob.globmatch(rel_path, self.config.patterns, root_dir=self.config.root, flags=self._glob_flags() | glob.REALPATH)

@dataclass
class GlobFileTrackerConfig:
    """Configuration for the glob file tracker.

    Raises:
        viat.exceptions.ViatConfigError: If some flag is unrecognized.
    """

    root: pathlib.Path
    """The root patch for matching relative patterns.

    Relative paths are resolved with respect to the vault root."""

    patterns: Sequence[str]
    """A sequence of wcmatch-flavored glob patterns."""

    flags: Sequence[str] = field(default_factory=lambda: DEFAULT_GLOB_FLAGS)
    """A sequence of strings corresponding to glob options."""

    # ruff: ignore[undocumented-magic-method]
    def __post_init__(self) -> None:
        validate_wcmatch_flags(self.flags)