Skip to content

viat.vault

A convenience module for vaults.

VIAT_SUBDIR = '.viat' module-attribute

The name of the default viat subdirectory.

ViatPathResolver

The path resolver for the ViatVault.

Parameters:

  • root (Path) –

    The root path of the vault.

    To find a vault root from a subdirectory, you can use the [locate][.locate] factory method instead.

Raises:

Source code in src/viat/_vault/resolver.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class ViatPathResolver:
    """The path resolver for the [`ViatVault`](viat.vault.ViatVault).

    Args:
        root: The root path of the vault.

            To find a vault root from a subdirectory, you can use the [`locate`][.locate] factory method instead.

    Raises:
        viat.exceptions.ViatConfigError: If the vault is misconfigured.
    """

    _root: pathlib.Path

    def __init__(self, root: pathlib.Path) -> None:
        self._root = root

    def get_root(self) -> pathlib.Path:
        """The root directory of the vault."""
        return self._root

    def get_viat(self) -> pathlib.Path:
        """The root directory of the vault."""
        return self._root / VIAT_SUBDIR

    def get_config(self, ext: str) -> pathlib.Path:
        """The root directory of the vault."""
        return self._root / VIAT_SUBDIR / f'config.{ext}'

    def relativize(self, path: pathlib.Path | str) -> pathlib.Path:
        """Relativize a path so that it can be used with an attribute storage.

        Args:
            path: The path to relativize. If it is already relative, it is left as-is.
        """
        try:
            return pathlib.Path(path).relative_to(self._root)
        except ValueError:
            return pathlib.Path(path)

get_config(ext)

The root directory of the vault.

Source code in src/viat/_vault/resolver.py 
33
34
35
def get_config(self, ext: str) -> pathlib.Path:
    """The root directory of the vault."""
    return self._root / VIAT_SUBDIR / f'config.{ext}'

get_root()

The root directory of the vault.

Source code in src/viat/_vault/resolver.py 
25
26
27
def get_root(self) -> pathlib.Path:
    """The root directory of the vault."""
    return self._root

get_viat()

The root directory of the vault.

Source code in src/viat/_vault/resolver.py 
29
30
31
def get_viat(self) -> pathlib.Path:
    """The root directory of the vault."""
    return self._root / VIAT_SUBDIR

relativize(path)

Relativize a path so that it can be used with an attribute storage.

Parameters:

  • path (Path | str) –

    The path to relativize. If it is already relative, it is left as-is.

Source code in src/viat/_vault/resolver.py 
37
38
39
40
41
42
43
44
45
46
def relativize(self, path: pathlib.Path | str) -> pathlib.Path:
    """Relativize a path so that it can be used with an attribute storage.

    Args:
        path: The path to relativize. If it is already relative, it is left as-is.
    """
    try:
        return pathlib.Path(path).relative_to(self._root)
    except ValueError:
        return pathlib.Path(path)

ViatVault

The default provider that stores all configuration and data in a .viat subdirectory.

Parameters:

  • root (Path) –

    The root path of the vault.

    To find a vault root from a subdirectory, you can use the helper [locate_existing_vault_root][.locate_existing_vault_root].

  • static_config (ViatVaultStaticConfig | None, default: None ) –

    Static configuration for the vault.

Raises:

Source code in src/viat/_vault/vault.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class ViatVault:
    """The default provider that stores all configuration and data in a `.viat` subdirectory.

    Args:
        root: The root path of the vault.

            To find a vault root from a subdirectory, you can use the helper
            [`locate_existing_vault_root`][.locate_existing_vault_root].

        static_config: Static configuration for the vault.

    Raises:
        viat.exceptions.ViatConfigError: If the vault is misconfigured.
    """

    resolver: ViatPathResolver
    """A resolver for paths related to the vault."""

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

    tracker: ViatFileTracker
    """The vault's tracker."""

    storage: ViatAttributeStorage
    """The vault's storage."""

    @staticmethod
    def initialize(root: pathlib.Path, static_config: ViatVaultStaticConfig | None = None) -> 'ViatVault':
        """Initialize a new vault.

        Args:
            root: The root of the new vault.
            static_config: Static configuration for the vault.

        Returns:
            A newly initialized vault

        Raises:
            viat.exceptions.ViatVaultError: If a vault already exists or if its creation fails.
        """
        resolver = ViatPathResolver(root)

        try:
            resolver.get_viat().mkdir(parents=True)
        except PermissionError as err:
            raise ViatVaultError('No permissions to initialize vault here') from err
        except FileExistsError as err:
            raise ViatVaultError('A vault has already been set up here') from err

        resolver.get_config('toml').touch()
        return ViatVault(root, static_config)

    def __init__(self, root: pathlib.Path, static_config: ViatVaultStaticConfig | None = None) -> None:
        self.resolver = ViatPathResolver(root)

        try:
            self.resolver.get_viat().stat()
        except FileNotFoundError as err:
            raise ViatVaultError(f'Missing {VIAT_SUBDIR} subdirectory') from err
        except OSError as err:
            raise ViatVaultError(f'The {VIAT_SUBDIR} subdirectory cannot be accessed') from err

        if config_loader := ConfigLoader.try_load_toml_file(self.resolver.get_config('toml')):
            if self.resolver.get_config('json').exists():
                raise ViatVaultError(f'Cannot have both static_config.toml and static_config.json in the {VIAT_SUBDIR} subdirectory')
        else:
            config_loader = ConfigLoader.try_load_json_file(self.resolver.get_config('json')) or ConfigLoader({})

        self.static_config = static_config or ViatVaultStaticConfig()
        self.tracker = load_tracker_from_config(self.resolver, self.static_config, config_loader)
        self.storage = load_storage_from_config(self.resolver, self.static_config, config_loader)

resolver = ViatPathResolver(root) instance-attribute

A resolver for paths related to the vault.

static_config = static_config or ViatVaultStaticConfig() instance-attribute

The vault's static configuration.

storage = load_storage_from_config(self.resolver, self.static_config, config_loader) instance-attribute

The vault's storage.

tracker = load_tracker_from_config(self.resolver, self.static_config, config_loader) instance-attribute

The vault's tracker.

initialize(root, static_config=None) staticmethod

Initialize a new vault.

Parameters:

  • root (Path) –

    The root of the new vault.

  • static_config (ViatVaultStaticConfig | None, default: None ) –

    Static configuration for the vault.

Returns:

Raises:

  • ViatVaultError –

    If a vault already exists or if its creation fails.

Source code in src/viat/_vault/vault.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@staticmethod
def initialize(root: pathlib.Path, static_config: ViatVaultStaticConfig | None = None) -> 'ViatVault':
    """Initialize a new vault.

    Args:
        root: The root of the new vault.
        static_config: Static configuration for the vault.

    Returns:
        A newly initialized vault

    Raises:
        viat.exceptions.ViatVaultError: If a vault already exists or if its creation fails.
    """
    resolver = ViatPathResolver(root)

    try:
        resolver.get_viat().mkdir(parents=True)
    except PermissionError as err:
        raise ViatVaultError('No permissions to initialize vault here') from err
    except FileExistsError as err:
        raise ViatVaultError('A vault has already been set up here') from err

    resolver.get_config('toml').touch()
    return ViatVault(root, static_config)

ViatVaultStaticConfig dataclass

Static (i.e. known before initialization) configuration for the vault.

Source code in src/viat/_vault/config.py 
4
5
6
7
8
9
@dataclass
class ViatVaultStaticConfig:
    """Static (i.e. known before initialization) configuration for the vault."""

    skip_validation: bool = False
    """Skip tracker and schema validation."""

skip_validation = False class-attribute instance-attribute

Skip tracker and schema validation.

autoload_vault(static_config=None)

Try to load a vault without any manual configuration.

Returns:

Raises:

Source code in src/viat/_vault/vault.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def autoload_vault(static_config: ViatVaultStaticConfig | None = None) -> ViatVault:
    """Try to load a vault without any manual configuration.

    Returns:
        An initialized vault.

    Raises:
        viat.exceptions.ViatCliError: If the loading fails.
    """
    try:
        return ViatVault(
            resolve_enforced_vault_path() or locate_existing_vault_root(pathlib.Path.cwd()),
            static_config=static_config,
        )
    except ViatVaultError as err:
        raise ViatVaultError('Could not load viat vault') from err

locate_existing_vault_root(base)

Traverse the file system until a vault is found.

Parameters:

  • base (Path) –

    The path to start the search at.

Returns:

  • Path –

    A path to an existing vault.

Raises:

Source code in src/viat/_vault/vault.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def locate_existing_vault_root(base: pathlib.Path) -> pathlib.Path:
    """Traverse the file system until a vault is found.

    Args:
        base: The path to start the search at.

    Returns:
        A path to an existing vault.

    Raises:
        viat.exceptions.ViatVaultError: If no vault is found.
    """
    candidate = base

    while not (candidate / VIAT_SUBDIR).exists() and candidate != candidate.parent:
        candidate = candidate.parent

    if (candidate / VIAT_SUBDIR).exists():
        return candidate

    raise ViatVaultError('Could not locate vault')

resolve_enforced_vault_path()

Try to resolve the path enforced using the VIAT_DIR environment variable.

Returns:

  • Path | None –

    Either a resolved path or None if the environment variable is not set.

Raises:

Source code in src/viat/_vault/vault.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def resolve_enforced_vault_path() -> pathlib.Path | None:
    """Try to resolve the path enforced using the VIAT_DIR environment variable.

    Returns:
        Either a resolved path or None if the environment variable is not set.

    Raises:
        viat.exceptions.ViatCliError: If the resolution fails.
    """
    if unresolved := os.environ.get('VIAT_DIR'):
        try:
            return pathlib.Path(unresolved).resolve()
        except OSError as err:
            raise ViatVaultError('Could not resolve path to viat vault') from err

    return None