Skip to content

Vault

The vault adapter provides integration with HashiCorp Vault for KV v2 secrets, dynamic credential leases, and transit encryption.

Ports

Abstract port interface defining the Vault adapter contract.

Port definitions for HashiCorp Vault operations.

archipy.adapters.vault.ports.VaultPort

Bases: ABC

Interface for HashiCorp Vault operations.

Defines the contract for Vault adapters covering KV v2 secrets, dynamic leases, and transit encryption.

Source code in archipy/adapters/vault/ports.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
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
class VaultPort(ABC):
    """Interface for HashiCorp Vault operations.

    Defines the contract for Vault adapters covering KV v2 secrets, dynamic
    leases, and transit encryption.
    """

    @abstractmethod
    def read_secret(self, path: str, mount_point: str | None = None) -> dict[str, Any]:
        """Read a KV v2 secret.

        Args:
            path: Secret path within the mount.
            mount_point: Optional KV v2 mount override; defaults to config mount.

        Returns:
            Secret key/value payload.
        """
        raise NotImplementedError

    @abstractmethod
    def write_secret(
        self,
        path: str,
        secret: dict[str, Any],
        mount_point: str | None = None,
    ) -> None:
        """Write (create or update) a KV v2 secret.

        Args:
            path: Secret path within the mount.
            secret: Key/value payload to store.
            mount_point: Optional KV v2 mount override; defaults to config mount.
        """
        raise NotImplementedError

    @abstractmethod
    def delete_secret(self, path: str, mount_point: str | None = None) -> None:
        """Delete a KV v2 secret (all versions and metadata).

        Args:
            path: Secret path within the mount.
            mount_point: Optional KV v2 mount override; defaults to config mount.
        """
        raise NotImplementedError

    @abstractmethod
    def list_secrets(self, path: str, mount_point: str | None = None) -> list[str]:
        """List secret keys under a path.

        Args:
            path: Path prefix to list.
            mount_point: Optional KV v2 mount override; defaults to config mount.

        Returns:
            List of key names (directories end with ``/``).
        """
        raise NotImplementedError

    @abstractmethod
    def get_dynamic_credentials(
        self,
        mount_point: str,
        role: str,
        *,
        parameters: dict[str, Any] | None = None,
    ) -> VaultLeaseDTO:
        """Generate dynamic credentials for a secrets-engine role.

        Args:
            mount_point: Secrets engine mount (e.g. ``database`` or ``ssh``).
            role: Role name that generates credentials.
            parameters: Optional engine-specific parameters (e.g. ``{"ip": "127.0.0.1"}``
                for SSH OTP). When not ``None``, issues a write to ``{mount}/creds/{role}``.
                When ``None``, uses the database secrets engine generate-credentials API
                (with a generic read fallback for other engines).

        Returns:
            Lease metadata and credential payload.
        """
        raise NotImplementedError

    @abstractmethod
    def renew_lease(self, lease_id: str, increment: int | None = None) -> VaultLeaseDTO:
        """Renew a dynamic secret lease.

        Args:
            lease_id: Lease identifier to renew.
            increment: Optional requested TTL extension in seconds.

        Returns:
            Updated lease metadata.
        """
        raise NotImplementedError

    @abstractmethod
    def revoke_lease(self, lease_id: str) -> None:
        """Revoke a dynamic secret lease.

        Args:
            lease_id: Lease identifier to revoke.
        """
        raise NotImplementedError

    @abstractmethod
    def encrypt(self, key_name: str, plaintext: str, mount_point: str = "transit") -> str:
        """Encrypt plaintext using the transit secrets engine.

        Args:
            key_name: Transit key name.
            plaintext: UTF-8 plaintext to encrypt.
            mount_point: Transit mount point.

        Returns:
            Vault ciphertext string (e.g. ``vault:v1:...``).
        """
        raise NotImplementedError

    @abstractmethod
    def decrypt(self, key_name: str, ciphertext: str, mount_point: str = "transit") -> str:
        """Decrypt ciphertext using the transit secrets engine.

        Args:
            key_name: Transit key name.
            ciphertext: Vault ciphertext string.
            mount_point: Transit mount point.

        Returns:
            Decoded UTF-8 plaintext.
        """
        raise NotImplementedError

archipy.adapters.vault.ports.VaultPort.read_secret abstractmethod 

read_secret(
    path: str, mount_point: str | None = None
) -> dict[str, Any]

Read a KV v2 secret.

Parameters:

Name Type Description Default
path str

Secret path within the mount.

 required
mount_point str | None

Optional KV v2 mount override; defaults to config mount.

 None

Returns:

Type Description
dict[str, Any]

Secret key/value payload.
Source code in archipy/adapters/vault/ports.py 
19
20
21
22
23
24
25
26
27
28
29
30
@abstractmethod
def read_secret(self, path: str, mount_point: str | None = None) -> dict[str, Any]:
    """Read a KV v2 secret.

    Args:
        path: Secret path within the mount.
        mount_point: Optional KV v2 mount override; defaults to config mount.

    Returns:
        Secret key/value payload.
    """
    raise NotImplementedError

archipy.adapters.vault.ports.VaultPort.write_secret abstractmethod 

write_secret(
    path: str,
    secret: dict[str, Any],
    mount_point: str | None = None,
) -> None

Write (create or update) a KV v2 secret.

Parameters:

Name Type Description Default
path str

Secret path within the mount.

 required
secret dict[str, Any]

Key/value payload to store.

 required
mount_point str | None

Optional KV v2 mount override; defaults to config mount.

 None
Source code in archipy/adapters/vault/ports.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@abstractmethod
def write_secret(
    self,
    path: str,
    secret: dict[str, Any],
    mount_point: str | None = None,
) -> None:
    """Write (create or update) a KV v2 secret.

    Args:
        path: Secret path within the mount.
        secret: Key/value payload to store.
        mount_point: Optional KV v2 mount override; defaults to config mount.
    """
    raise NotImplementedError

archipy.adapters.vault.ports.VaultPort.delete_secret abstractmethod 

delete_secret(
    path: str, mount_point: str | None = None
) -> None

Delete a KV v2 secret (all versions and metadata).

Parameters:

Name Type Description Default
path str

Secret path within the mount.

 required
mount_point str | None

Optional KV v2 mount override; defaults to config mount.

 None
Source code in archipy/adapters/vault/ports.py 
48
49
50
51
52
53
54
55
56
@abstractmethod
def delete_secret(self, path: str, mount_point: str | None = None) -> None:
    """Delete a KV v2 secret (all versions and metadata).

    Args:
        path: Secret path within the mount.
        mount_point: Optional KV v2 mount override; defaults to config mount.
    """
    raise NotImplementedError

archipy.adapters.vault.ports.VaultPort.list_secrets abstractmethod 

list_secrets(
    path: str, mount_point: str | None = None
) -> list[str]

List secret keys under a path.

Parameters:

Name Type Description Default
path str

Path prefix to list.

 required
mount_point str | None

Optional KV v2 mount override; defaults to config mount.

 None

Returns:

Type Description
list[str]

List of key names (directories end with /).
Source code in archipy/adapters/vault/ports.py 
58
59
60
61
62
63
64
65
66
67
68
69
@abstractmethod
def list_secrets(self, path: str, mount_point: str | None = None) -> list[str]:
    """List secret keys under a path.

    Args:
        path: Path prefix to list.
        mount_point: Optional KV v2 mount override; defaults to config mount.

    Returns:
        List of key names (directories end with ``/``).
    """
    raise NotImplementedError

archipy.adapters.vault.ports.VaultPort.get_dynamic_credentials abstractmethod 

get_dynamic_credentials(
    mount_point: str,
    role: str,
    *,
    parameters: dict[str, Any] | None = None,
) -> VaultLeaseDTO

Generate dynamic credentials for a secrets-engine role.

Parameters:

Name Type Description Default
mount_point str

Secrets engine mount (e.g. database or ssh).

 required
role str

Role name that generates credentials.

 required
parameters dict[str, Any] | None

Optional engine-specific parameters (e.g. {"ip": "127.0.0.1"} for SSH OTP). When not None, issues a write to {mount}/creds/{role}. When None, uses the database secrets engine generate-credentials API (with a generic read fallback for other engines).

 None

Returns:

Type Description
VaultLeaseDTO

Lease metadata and credential payload.
Source code in archipy/adapters/vault/ports.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@abstractmethod
def get_dynamic_credentials(
    self,
    mount_point: str,
    role: str,
    *,
    parameters: dict[str, Any] | None = None,
) -> VaultLeaseDTO:
    """Generate dynamic credentials for a secrets-engine role.

    Args:
        mount_point: Secrets engine mount (e.g. ``database`` or ``ssh``).
        role: Role name that generates credentials.
        parameters: Optional engine-specific parameters (e.g. ``{"ip": "127.0.0.1"}``
            for SSH OTP). When not ``None``, issues a write to ``{mount}/creds/{role}``.
            When ``None``, uses the database secrets engine generate-credentials API
            (with a generic read fallback for other engines).

    Returns:
        Lease metadata and credential payload.
    """
    raise NotImplementedError

archipy.adapters.vault.ports.VaultPort.renew_lease abstractmethod 

renew_lease(
    lease_id: str, increment: int | None = None
) -> VaultLeaseDTO

Renew a dynamic secret lease.

Parameters:

Name Type Description Default
lease_id str

Lease identifier to renew.

 required
increment int | None

Optional requested TTL extension in seconds.

 None

Returns:

Type Description
VaultLeaseDTO

Updated lease metadata.
Source code in archipy/adapters/vault/ports.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@abstractmethod
def renew_lease(self, lease_id: str, increment: int | None = None) -> VaultLeaseDTO:
    """Renew a dynamic secret lease.

    Args:
        lease_id: Lease identifier to renew.
        increment: Optional requested TTL extension in seconds.

    Returns:
        Updated lease metadata.
    """
    raise NotImplementedError

archipy.adapters.vault.ports.VaultPort.revoke_lease abstractmethod 

revoke_lease(lease_id: str) -> None

Revoke a dynamic secret lease.

Parameters:

Name Type Description Default
lease_id str

Lease identifier to revoke.

 required
Source code in archipy/adapters/vault/ports.py 
107
108
109
110
111
112
113
114
@abstractmethod
def revoke_lease(self, lease_id: str) -> None:
    """Revoke a dynamic secret lease.

    Args:
        lease_id: Lease identifier to revoke.
    """
    raise NotImplementedError

archipy.adapters.vault.ports.VaultPort.encrypt abstractmethod 

encrypt(
    key_name: str,
    plaintext: str,
    mount_point: str = "transit",
) -> str

Encrypt plaintext using the transit secrets engine.

Parameters:

Name Type Description Default
key_name str

Transit key name.

 required
plaintext str

UTF-8 plaintext to encrypt.

 required
mount_point str

Transit mount point.

 'transit'

Returns:

Type Description
str

Vault ciphertext string (e.g. vault:v1:...).
Source code in archipy/adapters/vault/ports.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
@abstractmethod
def encrypt(self, key_name: str, plaintext: str, mount_point: str = "transit") -> str:
    """Encrypt plaintext using the transit secrets engine.

    Args:
        key_name: Transit key name.
        plaintext: UTF-8 plaintext to encrypt.
        mount_point: Transit mount point.

    Returns:
        Vault ciphertext string (e.g. ``vault:v1:...``).
    """
    raise NotImplementedError

archipy.adapters.vault.ports.VaultPort.decrypt abstractmethod 

decrypt(
    key_name: str,
    ciphertext: str,
    mount_point: str = "transit",
) -> str

Decrypt ciphertext using the transit secrets engine.

Parameters:

Name Type Description Default
key_name str

Transit key name.

 required
ciphertext str

Vault ciphertext string.

 required
mount_point str

Transit mount point.

 'transit'

Returns:

Type Description
str

Decoded UTF-8 plaintext.
Source code in archipy/adapters/vault/ports.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
@abstractmethod
def decrypt(self, key_name: str, ciphertext: str, mount_point: str = "transit") -> str:
    """Decrypt ciphertext using the transit secrets engine.

    Args:
        key_name: Transit key name.
        ciphertext: Vault ciphertext string.
        mount_point: Transit mount point.

    Returns:
        Decoded UTF-8 plaintext.
    """
    raise NotImplementedError

options: show_root_toc_entry: false heading_level: 3

Adapters

Concrete Vault adapter wrapping hvac with ArchiPy conventions.

HashiCorp Vault adapter implementation using hvac.

archipy.adapters.vault.adapters.logger module-attribute 

logger = logging.getLogger(__name__)

archipy.adapters.vault.adapters.VaultExceptionHandlerMixin

Map hvac/network exceptions to ArchiPy domain errors.

Source code in archipy/adapters/vault/adapters.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class VaultExceptionHandlerMixin:
    """Map hvac/network exceptions to ArchiPy domain errors."""

    @classmethod
    def _handle_vault_exception(cls, exception: Exception, operation: str) -> NoReturn:
        """Convert Vault client failures into domain errors.

        Args:
            exception: Original exception from hvac or requests.
            operation: Name of the failing operation.

        Raises:
            NotFoundError: When the path or secret does not exist.
            PermissionDeniedError: When authz fails.
            InvalidArgumentError: When Vault rejects the request parameters.
            UnavailableError: When Vault is down or rate-limited.
            NetworkError: When the network call fails.
            ConfigurationError: For other Vault errors.
        """
        if isinstance(exception, hvac.exceptions.InvalidPath):
            raise NotFoundError(resource_type="vault_secret") from exception
        if isinstance(exception, hvac.exceptions.Forbidden | hvac.exceptions.Unauthorized):
            raise PermissionDeniedError(
                additional_data={"details": f"Permission denied for Vault operation: {operation}"},
            ) from exception
        if isinstance(exception, hvac.exceptions.InvalidRequest | hvac.exceptions.ParamValidationError):
            raise InvalidArgumentError(argument_name=operation) from exception
        if isinstance(exception, hvac.exceptions.VaultDown | hvac.exceptions.RateLimitExceeded):
            raise UnavailableError(resource_type="Vault", additional_data={"operation": operation}) from exception
        if isinstance(exception, RequestsConnectionError | RequestsTimeout):
            raise NetworkError(service="Vault") from exception
        if isinstance(exception, hvac.exceptions.VaultError):
            raise ConfigurationError(operation=operation, reason=str(exception)) from exception
        raise ConfigurationError(operation=operation, reason=str(exception)) from exception

archipy.adapters.vault.adapters.VaultAdapter

Bases: VaultPort, VaultExceptionHandlerMixin

Concrete Vault adapter wrapping hvac.Client.

Source code in archipy/adapters/vault/adapters.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
class VaultAdapter(VaultPort, VaultExceptionHandlerMixin):
    """Concrete Vault adapter wrapping ``hvac.Client``."""

    def __init__(self, vault_configs: VaultConfig | None = None) -> None:
        """Initialize the Vault adapter.

        Args:
            vault_configs: Optional Vault configuration. If None, uses
                ``BaseConfig.global_config().VAULT``.

        Raises:
            ConfigurationError: If authentication or connection fails.
            InvalidArgumentError: If required configuration is missing.
        """
        if vault_configs is not None:
            self.configs = vault_configs
        else:
            global_config = BaseConfig.global_config()
            vault_config = getattr(global_config, "VAULT", None)
            if not isinstance(vault_config, VaultConfig):
                raise InvalidArgumentError(argument_name="VAULT")
            self.configs = vault_config

        if not self.configs.ADDR:
            raise InvalidArgumentError(argument_name="ADDR")

        try:
            self._client = create_vault_client(self.configs)
        except ConfigurationError:
            raise
        except (hvac.exceptions.VaultError, RequestsConnectionError, RequestsTimeout, OSError, ValueError) as e:
            raise ConfigurationError(operation="vault_connect", reason=str(e)) from e

        self.token_renew_count = 0

        # Per-instance TTL cache so SECRET_CACHE_TTL on this config is honored.
        # Class-level decoration cannot see instance config and resolves TTL too early.
        ttl = max(self.configs.SECRET_CACHE_TTL, 0)
        self._read_secret_cached: Any | None = None
        if ttl > 0:
            cached = ttl_cache_decorator(ttl_seconds=ttl, maxsize=100)(VaultAdapter._read_secret_uncached)
            self._read_secret_cached = cached.__get__(self, VaultAdapter)

    def _mount(self, mount_point: str | None) -> str:
        """Resolve the KV mount point."""
        return mount_point if mount_point is not None else self.configs.MOUNT_POINT

    def _run[T](self, operation: str, fn: Callable[[], T]) -> T:
        """Execute a Vault call, mapping client failures to domain errors.

        Args:
            operation: Operation name for error context.
            fn: Zero-arg callable performing the Vault client work.

        Returns:
            Whatever ``fn`` returns.

        Raises:
            BaseError: Domain errors raised by ``fn`` are re-raised unchanged.
            ConfigurationError: For unmapped Vault/client failures.
        """
        self._maybe_renew_token()
        try:
            return fn()
        except BaseError:
            raise
        except (hvac.exceptions.VaultError, RequestsConnectionError, RequestsTimeout, OSError, ValueError) as e:
            self._handle_vault_exception(e, operation)

    def _maybe_renew_token(self) -> None:
        """Renew the client token when AUTO_RENEW_TOKEN is enabled and TTL is low."""
        if not self.configs.AUTO_RENEW_TOKEN:
            return
        try:
            lookup = self._client.auth.token.lookup_self()
            ttl = int(lookup.get("data", {}).get("ttl", 0))
            if ttl <= self.configs.RENEW_THRESHOLD_SECONDS:
                self._client.auth.token.renew_self()
                self.token_renew_count += 1
                logger.debug("Renewed Vault token (previous TTL=%s)", ttl)
        except (hvac.exceptions.VaultError, RequestsConnectionError, RequestsTimeout, OSError, ValueError) as e:
            self._handle_vault_exception(e, "renew_token")

    def clear_secret_cache(self) -> None:
        """Clear all cached ``read_secret`` results."""
        cached = self._read_secret_cached
        if cached is not None and hasattr(cached, "clear_cache"):
            cached.clear_cache()

    def _read_secret_uncached(self, path: str, mount_point: str | None = None) -> dict[str, Any]:
        """Fetch a KV v2 secret without using the instance cache."""
        mount = self._mount(mount_point)

        def _read() -> dict[str, Any]:
            response = self._client.secrets.kv.v2.read_secret_version(
                path=path,
                mount_point=mount,
                raise_on_deleted_version=True,
            )
            data = response.get("data", {}).get("data") or {}
            if not isinstance(data, dict):
                raise ConfigurationError(
                    operation="read_secret",
                    reason=f"Vault secret at '{path}' did not return a dict payload",
                )
            return {str(k): v for k, v in data.items()}

        return self._run("read_secret", _read)

    @override
    def read_secret(self, path: str, mount_point: str | None = None) -> dict[str, Any]:
        """Read a KV v2 secret.

        Args:
            path: Secret path within the mount.
            mount_point: Optional KV v2 mount override.

        Returns:
            Secret key/value payload.

        Raises:
            InvalidArgumentError: If path is empty.
            NotFoundError: If the secret does not exist.
            PermissionDeniedError: If access is denied.
            ConfigurationError: For other Vault errors.
        """
        if not path:
            raise InvalidArgumentError(argument_name="path")
        if self._read_secret_cached is not None:
            return self._read_secret_cached(path, mount_point)
        return self._read_secret_uncached(path, mount_point)

    @override
    def write_secret(
        self,
        path: str,
        secret: dict[str, Any],
        mount_point: str | None = None,
    ) -> None:
        """Write a KV v2 secret.

        Args:
            path: Secret path within the mount.
            secret: Key/value payload to store.
            mount_point: Optional KV v2 mount override.

        Raises:
            InvalidArgumentError: If path is empty or secret is empty.
            PermissionDeniedError: If access is denied.
            ConfigurationError: For other Vault errors.
        """
        if not path:
            raise InvalidArgumentError(argument_name="path")
        if not secret:
            raise InvalidArgumentError(argument_name="secret")
        mount = self._mount(mount_point)

        def _write() -> None:
            self._client.secrets.kv.v2.create_or_update_secret(
                path=path,
                secret=secret,
                mount_point=mount,
            )
            self.clear_secret_cache()

        self._run("write_secret", _write)

    @override
    def delete_secret(self, path: str, mount_point: str | None = None) -> None:
        """Delete a KV v2 secret including metadata and all versions.

        Args:
            path: Secret path within the mount.
            mount_point: Optional KV v2 mount override.

        Raises:
            InvalidArgumentError: If path is empty.
            NotFoundError: If the secret does not exist.
            PermissionDeniedError: If access is denied.
            ConfigurationError: For other Vault errors.
        """
        if not path:
            raise InvalidArgumentError(argument_name="path")
        mount = self._mount(mount_point)

        def _delete() -> None:
            self._client.secrets.kv.v2.delete_metadata_and_all_versions(
                path=path,
                mount_point=mount,
            )
            self.clear_secret_cache()

        self._run("delete_secret", _delete)

    @override
    def list_secrets(self, path: str, mount_point: str | None = None) -> list[str]:
        """List secret keys under a path.

        Args:
            path: Path prefix to list.
            mount_point: Optional KV v2 mount override.

        Returns:
            List of key names.

        Raises:
            NotFoundError: If the path does not exist.
            PermissionDeniedError: If access is denied.
            ConfigurationError: For other Vault errors.
        """
        mount = self._mount(mount_point)

        def _list() -> list[str]:
            response = self._client.secrets.kv.v2.list_secrets(path=path, mount_point=mount)
            keys = response.get("data", {}).get("keys") or []
            return [str(key) for key in keys]

        return self._run("list_secrets", _list)

    @override
    def get_dynamic_credentials(
        self,
        mount_point: str,
        role: str,
        *,
        parameters: dict[str, Any] | None = None,
    ) -> VaultLeaseDTO:
        """Generate dynamic credentials for a secrets-engine role.

        Args:
            mount_point: Secrets engine mount (e.g. ``database`` or ``ssh``).
            role: Role name that generates credentials.
            parameters: Optional engine-specific parameters. When provided (including
                an empty dict), issues a write to ``{mount}/creds/{role}`` (needed for
                engines like SSH OTP). When ``None``, uses the typed database API
                ``secrets.database.generate_credentials``, falling back to a generic
                read for non-database engines.

        Returns:
            Lease metadata and credential payload.

        Raises:
            InvalidArgumentError: If mount_point or role is empty.
            PermissionDeniedError: If access is denied.
            ConfigurationError: For other Vault errors.
        """
        if not mount_point:
            raise InvalidArgumentError(argument_name="mount_point")
        if not role:
            raise InvalidArgumentError(argument_name="role")
        path = f"{mount_point}/creds/{role}"

        def _get_creds() -> VaultLeaseDTO:
            if parameters is not None:
                response = self._client.write(path, **parameters)
            else:
                try:
                    response = self._client.secrets.database.generate_credentials(
                        name=role,
                        mount_point=mount_point,
                    )
                except hvac.exceptions.InvalidPath, hvac.exceptions.UnexpectedError:
                    response = self._client.read(path)
            if response is None:
                raise NotFoundError(resource_type="vault_role")
            return _lease_from_response(response, include_credential_data=True)

        return self._run("get_dynamic_credentials", _get_creds)

    @override
    def renew_lease(self, lease_id: str, increment: int | None = None) -> VaultLeaseDTO:
        """Renew a dynamic secret lease.

        Args:
            lease_id: Lease identifier to renew.
            increment: Optional requested TTL extension in seconds.

        Returns:
            Updated lease metadata.

        Raises:
            InvalidArgumentError: If lease_id is empty.
            ConfigurationError: For Vault errors.
        """
        if not lease_id:
            raise InvalidArgumentError(argument_name="lease_id")

        def _renew() -> VaultLeaseDTO:
            kwargs: dict[str, Any] = {"lease_id": lease_id}
            if increment is not None:
                kwargs["increment"] = increment
            response = self._client.sys.renew_lease(**kwargs)
            return _lease_from_response(
                response,
                default_lease_id=lease_id,
                include_credential_data=False,
            )

        return self._run("renew_lease", _renew)

    @override
    def revoke_lease(self, lease_id: str) -> None:
        """Revoke a dynamic secret lease.

        Args:
            lease_id: Lease identifier to revoke.

        Raises:
            InvalidArgumentError: If lease_id is empty.
            ConfigurationError: For Vault errors.
        """
        if not lease_id:
            raise InvalidArgumentError(argument_name="lease_id")
        self._run("revoke_lease", lambda: self._client.sys.revoke_lease(lease_id=lease_id))

    @override
    def encrypt(self, key_name: str, plaintext: str, mount_point: str = "transit") -> str:
        """Encrypt plaintext using the transit secrets engine.

        Args:
            key_name: Transit key name.
            plaintext: UTF-8 plaintext to encrypt.
            mount_point: Transit mount point.

        Returns:
            Vault ciphertext string.

        Raises:
            InvalidArgumentError: If key_name or plaintext is empty.
            ConfigurationError: For Vault errors.
        """
        if not key_name:
            raise InvalidArgumentError(argument_name="key_name")
        if plaintext == "":
            raise InvalidArgumentError(argument_name="plaintext")
        encoded = base64.b64encode(plaintext.encode("utf-8")).decode("ascii")

        def _encrypt() -> str:
            response = self._client.secrets.transit.encrypt_data(
                name=key_name,
                plaintext=encoded,
                mount_point=mount_point,
            )
            return str(response["data"]["ciphertext"])

        return self._run("encrypt", _encrypt)

    @override
    def decrypt(self, key_name: str, ciphertext: str, mount_point: str = "transit") -> str:
        """Decrypt ciphertext using the transit secrets engine.

        Args:
            key_name: Transit key name.
            ciphertext: Vault ciphertext string.
            mount_point: Transit mount point.

        Returns:
            Decoded UTF-8 plaintext.

        Raises:
            InvalidArgumentError: If key_name or ciphertext is empty.
            ConfigurationError: For Vault errors.
        """
        if not key_name:
            raise InvalidArgumentError(argument_name="key_name")
        if not ciphertext:
            raise InvalidArgumentError(argument_name="ciphertext")

        def _decrypt() -> str:
            response = self._client.secrets.transit.decrypt_data(
                name=key_name,
                ciphertext=ciphertext,
                mount_point=mount_point,
            )
            encoded = str(response["data"]["plaintext"])
            return base64.b64decode(encoded.encode("ascii")).decode("utf-8")

        return self._run("decrypt", _decrypt)

archipy.adapters.vault.adapters.VaultAdapter.configs instance-attribute 

configs = vault_configs

archipy.adapters.vault.adapters.VaultAdapter.token_renew_count instance-attribute 

token_renew_count = 0

archipy.adapters.vault.adapters.VaultAdapter.clear_secret_cache 

clear_secret_cache() -> None

Clear all cached read_secret results.

Source code in archipy/adapters/vault/adapters.py 
184
185
186
187
188
def clear_secret_cache(self) -> None:
    """Clear all cached ``read_secret`` results."""
    cached = self._read_secret_cached
    if cached is not None and hasattr(cached, "clear_cache"):
        cached.clear_cache()

archipy.adapters.vault.adapters.VaultAdapter.read_secret 

read_secret(
    path: str, mount_point: str | None = None
) -> dict[str, Any]

Read a KV v2 secret.

Parameters:

Name Type Description Default
path str

Secret path within the mount.

 required
mount_point str | None

Optional KV v2 mount override.

 None

Returns:

Type Description
dict[str, Any]

Secret key/value payload.

Raises:

Type Description
InvalidArgumentError

If path is empty.
NotFoundError

If the secret does not exist.
PermissionDeniedError

If access is denied.
ConfigurationError

For other Vault errors.
Source code in archipy/adapters/vault/adapters.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
@override
def read_secret(self, path: str, mount_point: str | None = None) -> dict[str, Any]:
    """Read a KV v2 secret.

    Args:
        path: Secret path within the mount.
        mount_point: Optional KV v2 mount override.

    Returns:
        Secret key/value payload.

    Raises:
        InvalidArgumentError: If path is empty.
        NotFoundError: If the secret does not exist.
        PermissionDeniedError: If access is denied.
        ConfigurationError: For other Vault errors.
    """
    if not path:
        raise InvalidArgumentError(argument_name="path")
    if self._read_secret_cached is not None:
        return self._read_secret_cached(path, mount_point)
    return self._read_secret_uncached(path, mount_point)

archipy.adapters.vault.adapters.VaultAdapter.write_secret 

write_secret(
    path: str,
    secret: dict[str, Any],
    mount_point: str | None = None,
) -> None

Write a KV v2 secret.

Parameters:

Name Type Description Default
path str

Secret path within the mount.

 required
secret dict[str, Any]

Key/value payload to store.

 required
mount_point str | None

Optional KV v2 mount override.

 None

Raises:

Type Description
InvalidArgumentError

If path is empty or secret is empty.
PermissionDeniedError

If access is denied.
ConfigurationError

For other Vault errors.
Source code in archipy/adapters/vault/adapters.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
@override
def write_secret(
    self,
    path: str,
    secret: dict[str, Any],
    mount_point: str | None = None,
) -> None:
    """Write a KV v2 secret.

    Args:
        path: Secret path within the mount.
        secret: Key/value payload to store.
        mount_point: Optional KV v2 mount override.

    Raises:
        InvalidArgumentError: If path is empty or secret is empty.
        PermissionDeniedError: If access is denied.
        ConfigurationError: For other Vault errors.
    """
    if not path:
        raise InvalidArgumentError(argument_name="path")
    if not secret:
        raise InvalidArgumentError(argument_name="secret")
    mount = self._mount(mount_point)

    def _write() -> None:
        self._client.secrets.kv.v2.create_or_update_secret(
            path=path,
            secret=secret,
            mount_point=mount,
        )
        self.clear_secret_cache()

    self._run("write_secret", _write)

archipy.adapters.vault.adapters.VaultAdapter.delete_secret 

delete_secret(
    path: str, mount_point: str | None = None
) -> None

Delete a KV v2 secret including metadata and all versions.

Parameters:

Name Type Description Default
path str

Secret path within the mount.

 required
mount_point str | None

Optional KV v2 mount override.

 None

Raises:

Type Description
InvalidArgumentError

If path is empty.
NotFoundError

If the secret does not exist.
PermissionDeniedError

If access is denied.
ConfigurationError

For other Vault errors.
Source code in archipy/adapters/vault/adapters.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
@override
def delete_secret(self, path: str, mount_point: str | None = None) -> None:
    """Delete a KV v2 secret including metadata and all versions.

    Args:
        path: Secret path within the mount.
        mount_point: Optional KV v2 mount override.

    Raises:
        InvalidArgumentError: If path is empty.
        NotFoundError: If the secret does not exist.
        PermissionDeniedError: If access is denied.
        ConfigurationError: For other Vault errors.
    """
    if not path:
        raise InvalidArgumentError(argument_name="path")
    mount = self._mount(mount_point)

    def _delete() -> None:
        self._client.secrets.kv.v2.delete_metadata_and_all_versions(
            path=path,
            mount_point=mount,
        )
        self.clear_secret_cache()

    self._run("delete_secret", _delete)

archipy.adapters.vault.adapters.VaultAdapter.list_secrets 

list_secrets(
    path: str, mount_point: str | None = None
) -> list[str]

List secret keys under a path.

Parameters:

Name Type Description Default
path str

Path prefix to list.

 required
mount_point str | None

Optional KV v2 mount override.

 None

Returns:

Type Description
list[str]

List of key names.

Raises:

Type Description
NotFoundError

If the path does not exist.
PermissionDeniedError

If access is denied.
ConfigurationError

For other Vault errors.
Source code in archipy/adapters/vault/adapters.py 
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
@override
def list_secrets(self, path: str, mount_point: str | None = None) -> list[str]:
    """List secret keys under a path.

    Args:
        path: Path prefix to list.
        mount_point: Optional KV v2 mount override.

    Returns:
        List of key names.

    Raises:
        NotFoundError: If the path does not exist.
        PermissionDeniedError: If access is denied.
        ConfigurationError: For other Vault errors.
    """
    mount = self._mount(mount_point)

    def _list() -> list[str]:
        response = self._client.secrets.kv.v2.list_secrets(path=path, mount_point=mount)
        keys = response.get("data", {}).get("keys") or []
        return [str(key) for key in keys]

    return self._run("list_secrets", _list)

archipy.adapters.vault.adapters.VaultAdapter.get_dynamic_credentials 

get_dynamic_credentials(
    mount_point: str,
    role: str,
    *,
    parameters: dict[str, Any] | None = None,
) -> VaultLeaseDTO

Generate dynamic credentials for a secrets-engine role.

Parameters:

Name Type Description Default
mount_point str

Secrets engine mount (e.g. database or ssh).

 required
role str

Role name that generates credentials.

 required
parameters dict[str, Any] | None

Optional engine-specific parameters. When provided (including an empty dict), issues a write to {mount}/creds/{role} (needed for engines like SSH OTP). When None, uses the typed database API secrets.database.generate_credentials, falling back to a generic read for non-database engines.

 None

Returns:

Type Description
VaultLeaseDTO

Lease metadata and credential payload.

Raises:

Type Description
InvalidArgumentError

If mount_point or role is empty.
PermissionDeniedError

If access is denied.
ConfigurationError

For other Vault errors.
Source code in archipy/adapters/vault/adapters.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
@override
def get_dynamic_credentials(
    self,
    mount_point: str,
    role: str,
    *,
    parameters: dict[str, Any] | None = None,
) -> VaultLeaseDTO:
    """Generate dynamic credentials for a secrets-engine role.

    Args:
        mount_point: Secrets engine mount (e.g. ``database`` or ``ssh``).
        role: Role name that generates credentials.
        parameters: Optional engine-specific parameters. When provided (including
            an empty dict), issues a write to ``{mount}/creds/{role}`` (needed for
            engines like SSH OTP). When ``None``, uses the typed database API
            ``secrets.database.generate_credentials``, falling back to a generic
            read for non-database engines.

    Returns:
        Lease metadata and credential payload.

    Raises:
        InvalidArgumentError: If mount_point or role is empty.
        PermissionDeniedError: If access is denied.
        ConfigurationError: For other Vault errors.
    """
    if not mount_point:
        raise InvalidArgumentError(argument_name="mount_point")
    if not role:
        raise InvalidArgumentError(argument_name="role")
    path = f"{mount_point}/creds/{role}"

    def _get_creds() -> VaultLeaseDTO:
        if parameters is not None:
            response = self._client.write(path, **parameters)
        else:
            try:
                response = self._client.secrets.database.generate_credentials(
                    name=role,
                    mount_point=mount_point,
                )
            except hvac.exceptions.InvalidPath, hvac.exceptions.UnexpectedError:
                response = self._client.read(path)
        if response is None:
            raise NotFoundError(resource_type="vault_role")
        return _lease_from_response(response, include_credential_data=True)

    return self._run("get_dynamic_credentials", _get_creds)

archipy.adapters.vault.adapters.VaultAdapter.renew_lease 

renew_lease(
    lease_id: str, increment: int | None = None
) -> VaultLeaseDTO

Renew a dynamic secret lease.

Parameters:

Name Type Description Default
lease_id str

Lease identifier to renew.

 required
increment int | None

Optional requested TTL extension in seconds.

 None

Returns:

Type Description
VaultLeaseDTO

Updated lease metadata.

Raises:

Type Description
InvalidArgumentError

If lease_id is empty.
ConfigurationError

For Vault errors.
Source code in archipy/adapters/vault/adapters.py 
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
@override
def renew_lease(self, lease_id: str, increment: int | None = None) -> VaultLeaseDTO:
    """Renew a dynamic secret lease.

    Args:
        lease_id: Lease identifier to renew.
        increment: Optional requested TTL extension in seconds.

    Returns:
        Updated lease metadata.

    Raises:
        InvalidArgumentError: If lease_id is empty.
        ConfigurationError: For Vault errors.
    """
    if not lease_id:
        raise InvalidArgumentError(argument_name="lease_id")

    def _renew() -> VaultLeaseDTO:
        kwargs: dict[str, Any] = {"lease_id": lease_id}
        if increment is not None:
            kwargs["increment"] = increment
        response = self._client.sys.renew_lease(**kwargs)
        return _lease_from_response(
            response,
            default_lease_id=lease_id,
            include_credential_data=False,
        )

    return self._run("renew_lease", _renew)

archipy.adapters.vault.adapters.VaultAdapter.revoke_lease 

revoke_lease(lease_id: str) -> None

Revoke a dynamic secret lease.

Parameters:

Name Type Description Default
lease_id str

Lease identifier to revoke.

 required

Raises:

Type Description
InvalidArgumentError

If lease_id is empty.
ConfigurationError

For Vault errors.
Source code in archipy/adapters/vault/adapters.py 
401
402
403
404
405
406
407
408
409
410
411
412
413
414
@override
def revoke_lease(self, lease_id: str) -> None:
    """Revoke a dynamic secret lease.

    Args:
        lease_id: Lease identifier to revoke.

    Raises:
        InvalidArgumentError: If lease_id is empty.
        ConfigurationError: For Vault errors.
    """
    if not lease_id:
        raise InvalidArgumentError(argument_name="lease_id")
    self._run("revoke_lease", lambda: self._client.sys.revoke_lease(lease_id=lease_id))

archipy.adapters.vault.adapters.VaultAdapter.encrypt 

encrypt(
    key_name: str,
    plaintext: str,
    mount_point: str = "transit",
) -> str

Encrypt plaintext using the transit secrets engine.

Parameters:

Name Type Description Default
key_name str

Transit key name.

 required
plaintext str

UTF-8 plaintext to encrypt.

 required
mount_point str

Transit mount point.

 'transit'

Returns:

Type Description
str

Vault ciphertext string.

Raises:

Type Description
InvalidArgumentError

If key_name or plaintext is empty.
ConfigurationError

For Vault errors.
Source code in archipy/adapters/vault/adapters.py 
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
@override
def encrypt(self, key_name: str, plaintext: str, mount_point: str = "transit") -> str:
    """Encrypt plaintext using the transit secrets engine.

    Args:
        key_name: Transit key name.
        plaintext: UTF-8 plaintext to encrypt.
        mount_point: Transit mount point.

    Returns:
        Vault ciphertext string.

    Raises:
        InvalidArgumentError: If key_name or plaintext is empty.
        ConfigurationError: For Vault errors.
    """
    if not key_name:
        raise InvalidArgumentError(argument_name="key_name")
    if plaintext == "":
        raise InvalidArgumentError(argument_name="plaintext")
    encoded = base64.b64encode(plaintext.encode("utf-8")).decode("ascii")

    def _encrypt() -> str:
        response = self._client.secrets.transit.encrypt_data(
            name=key_name,
            plaintext=encoded,
            mount_point=mount_point,
        )
        return str(response["data"]["ciphertext"])

    return self._run("encrypt", _encrypt)

archipy.adapters.vault.adapters.VaultAdapter.decrypt 

decrypt(
    key_name: str,
    ciphertext: str,
    mount_point: str = "transit",
) -> str

Decrypt ciphertext using the transit secrets engine.

Parameters:

Name Type Description Default
key_name str

Transit key name.

 required
ciphertext str

Vault ciphertext string.

 required
mount_point str

Transit mount point.

 'transit'

Returns:

Type Description
str

Decoded UTF-8 plaintext.

Raises:

Type Description
InvalidArgumentError

If key_name or ciphertext is empty.
ConfigurationError

For Vault errors.
Source code in archipy/adapters/vault/adapters.py 
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
@override
def decrypt(self, key_name: str, ciphertext: str, mount_point: str = "transit") -> str:
    """Decrypt ciphertext using the transit secrets engine.

    Args:
        key_name: Transit key name.
        ciphertext: Vault ciphertext string.
        mount_point: Transit mount point.

    Returns:
        Decoded UTF-8 plaintext.

    Raises:
        InvalidArgumentError: If key_name or ciphertext is empty.
        ConfigurationError: For Vault errors.
    """
    if not key_name:
        raise InvalidArgumentError(argument_name="key_name")
    if not ciphertext:
        raise InvalidArgumentError(argument_name="ciphertext")

    def _decrypt() -> str:
        response = self._client.secrets.transit.decrypt_data(
            name=key_name,
            ciphertext=ciphertext,
            mount_point=mount_point,
        )
        encoded = str(response["data"]["plaintext"])
        return base64.b64decode(encoded.encode("ascii")).decode("utf-8")

    return self._run("decrypt", _decrypt)

options: show_root_toc_entry: false heading_level: 3